边缘计算南向采集通道优化与设备画像构建方案 V2
文档状态（Q3 归档）： 本文为 2026 Q2 规划归档，保留历史设计细节供查阅。现行交付、差距跟踪与验收以 Q3 采集优化方案、edge/边缘网关架构设计总览 与 TODO/ScanEngine 重构方案 为准。
架构说明： 南向采集已迁移至 ScanEngine 调度驱动内核（
CollectionScheduler/deviceLoop已移除）。下文deviceLoop/collectDevice/CollectionScheduler等为 Q2 规划态描述，请勿作为现行实现依据。
1. 项目现状分析
1.1 当前架构
当前项目采用三级架构：
- 通道(Channel)：对应一个采集驱动，管理多个设备
- 设备(Device)：具体的设备实例，包含多个点位
- 点位(Point)：设备的具体数据点
核心组件：
ChannelManager：管理所有采集通道及其设备CommunicationManageTemplate：管理设备运行状态- 多协议驱动：支持Modbus、BACnet、OPC UA、S7等协议
1.2 现有功能
- 设备状态管理（在线、离线、不稳定等状态）
- 定时数据采集
- 基本的设备指标收集
- 多协议支持
- 设备扫描和点位发现
1.3 存在的问题
- 缺乏设备级别的智能适配机制
- 采集参数固定，无法根据设备特性动态调整
- 没有统一的设备画像结构
- 并发采集策略不够优化
- 批量采集效率有待提高
- 缺乏RTT自适应和MTU动态协商机制
- 心跳保活机制不完善，死链检测慢
- 固定MTU限制大数据块传输
- 地址分散导致请求碎片化
- 缺乏心跳保活机制，死链检测慢
- 异常恢复时间长
1.4 性能瓶颈分析与优化目标
| 维度 | 当前痛点 | 优化目标 | 关键技术 |
|---|---|---|---|
| 延迟 (Latency) | 静态超时/间隔策略导致长尾延迟；网络抖动引起重传风暴。 | 99th 百分位延迟 < 10ms | RTT 动态自适应窗口；EWMA 预测算法；QoS 优先级队列 |
| 吞吐 (Throughput) | 固定 MTU (如 Modbus 250B) 限制大数据块传输；地址分散导致请求碎片化。 | 自适应 MTU (最大支持 4096B) | 智能 MTU 动态协商；Gap 空洞合并算法；零拷贝环形缓冲 |
| 稳定性 (Stability) | 缺乏心跳保活机制，死链检测慢；异常恢复时间长。 | 故障检测 < 3s；自动重连恢复 | 智能业务心跳 (Traffic-Aware Heartbeat)；链路层状态机 |
1.5 分层架构重构原则
采用 南向采集 -> 缓存 -> 影子设备 -> 边缘计算 -> 北向上报 的分层处理，并引入影子设备作为唯一内部数据模型中心：
- 采集层 (Collection Layer):
- 驱动适配: 统一封装 Serial, TCP, UDP 等物理通道。
- 协议适配: 适配 Modbus, S7, BACnet 等工业协议，增加
HeartbeatInterval,BufferSize,Qos等控制参数。 - 优化策略: 实现 RTT 统计、MTU 协商状态机与 Gap 合并策略。
- 缓存层 (Buffer Layer):
- 使用零拷贝环形缓冲 (Ring Buffer) 存储原始报文。
- 实现 ACK 压缩与重复包过滤。
- 边缘计算层 (Edge Computing Layer):
- 执行 TSL 物模型解析、公式计算、数据清洗。
- 动态调整采集策略 (如降级、加速)。
- 影子设备核心层 (Shadow Core Layer):
- Real Shadow Store：纯内存运行时快照。
- Virtual Shadow Engine：跨设备公式依赖计算。
- Edge Compute Engine：差值/步长/滑动平均/阈值计算。
- Atomic Dispatcher：版本推进 + QoS ACK 机制。
- 北向上报层 (Northbound Layer):
- 数据标准化 JSON/MQTT 上报。
- 提供
COLLECT_CTRL指令通道，支持参数热更新(已实现简单的双向控制功能的基础上优化)。 - 统一鉴权、限流、审计机制。
2. 智能画像方案设计
2.1 设计目标
- 统一设备画像：为所有协议设备建立统一的画像结构
- 智能适配：根据设备特性自动调整采集策略
- 并发优化：合理安排采集顺序，提高并发效率
- 批量优化：优化同一总线上的设备采集，减少通信开销
- RTT自适应：基于往返时间动态调整采集参数
- MTU动态协商：智能调整最大传输单元，提高吞吐量
- 可扩展性：模块化设计，易于添加新协议支持
- 监控能力：提供设备状态监控和性能统计
2.2 可移植抽象层设计 (HAL)
定义统一的通信驱动接口 CommDriver，屏蔽底层差异：
type CommDriver interface {
Send(packet []byte) (timestamp time.Time, err error)
Recv() (packet []byte, timestamp time.Time, err error)
SetParam(key string, value interface{}) error // 设置 BufferSize, Timeout 等
GetParam(key string) (interface{}, error)
GetStats() CommStats // 获取丢包、RTT 等统计
}
2.3 关键参数清单
结合项目已有的协议扩展，以下为关键配置参数：
2.3.1 通用协议参数 (已集成)
针对所有支持的协议 (Modbus, TCP, S7, BACnet, OPC UA, Serial)，统一增加以下控制字段：
HeartbeatInterval(ms): 应用层心跳间隔。- 优化逻辑: 仅在链路空闲且超过该间隔时才发送保活包。
- 抑制策略: 任何成功的业务通信（如 Modbus 读取响应）都会重置心跳计时器，避免在正常采集时产生额外流量。
BufferSize(bytes): 接收缓冲区大小。配合 MTU 协商，支持大数据包接收 (默认 4096)。Qos(0-2): 服务质量等级。- 0: 尽力而为 (Fire and Forget)。
- 1: 至少一次 (At least once, 需要 ACK)。
- 2: 恰好一次 (Exactly once, 防重)。
Interval(ms): 采集轮询间隔 (部分协议如 BACnet/S7/OPC UA 新增/修正)。
2.3.2 RTT 模式参数 (自适应)
rtt_sample_window: 采样窗口大小 (默认 20)。rtt_alpha_gain: EWMA (指数加权移动平均) 增益系数 (默认 0.125)。rtt_threshold_high: 高延迟阈值 (触发降级/报警)。rtt_threshold_low: 低延迟阈值 (触发加速采集)。rtt_probe_interval_ms: RTT 探测包发送间隔。
2.3.3 智能 MTU 模式参数
mtu_max: 最大支持 MTU (如 Modbus TCP 250, S7 PDU 960)。mtu_min: 最小安全 MTU (如 64)。mtu_step: 探测步长 (二分法或线性递增)。mtu_hysteresis: 抖动滞后因子 (防止频繁切换)。
2.3.4 Gap 合并策略参数 (Batch Read Optimization)
针对 Modbus/PLC 等寄存器读取协议，通过合并相邻地址请求提升总线效率：
gap_max_hole: 最大允许地址空洞 (默认 0，表示完全连续才合并；最大受 MTU 限制)。gap_fill_strategy: 空洞填充策略 (0: 填充无用数据读取; 1: 分割请求)。gap_dynamic_enable: 是否启用基于 RTT 的动态 Gap 调整 (RTT 低时增大 Gap，RTT 高时减小 Gap)。
2.4 核心结构体定义
2.4.1 采集通信上下文
管理单条连接的生命周期与状态。
type CollectCommContext struct {
RTTBuffer []RTTNode // RTT 统计环缓冲
RTTIndex int // 环缓冲当前索引
MTUState *MTUFSM // MTU 协商状态机
BatchRead *BatchReadSnapshot // Gap 合并策略快照
LastActivity time.Time // 最后活跃时间 (业务通信成功时刻)
Stats CommStats // 统计信息
Config ProtocolConfig // 协议配置 (含 Heartbeat, BufferSize 等)
}
type ProtocolConfig struct {
HeartbeatInterval int
BufferSize int
Qos int
Interval int
}
2.4.2 RTT 统计节点
用于计算单次通信质量。
type RTTNode struct {
SeqNo uint16 // 序列号
SendTS time.Time // 发送时间戳
RecvTS time.Time // 接收时间戳
AckStatus bool // 是否收到 ACK
RTT int64 // 计算出的 RTT (us)
}
2.4.3 MTU 协商记录
记录 MTU 探测历史，辅助状态机决策。
type MTUNegotiationRecord struct {
AttemptValue int // 尝试值
ResponseTime int64 // 耗时 (us)
RetryCount int // 重试次数
Success bool // 结果
Timestamp time.Time // 记录时间
}
2.4.4 Batch Read 快照
记录当前 Gap 合并策略的运行状态。
type BatchReadSnapshot struct {
CurrentGap int // 当前生效的 Gap 值
MaxGap int // 允许的最大 Gap (受 MTU 限制)
MergedRequests uint64 // 合并后的请求总数
SavedRequests uint64 // 节省的请求数量 (原请求数 - 合并后请求数)
FillEfficiency float64 // 填充效率 (有效数据 / 总请求数据)
}
2.5 核心设计
2.5.1 统一设备画像结构
// 设备通信画像结构
type DeviceCommunicationProfile struct {
DeviceID string // 设备唯一标识
ChannelID string // 所属通道ID
ProtocolType string // 协议类型
SlaveID interface{} // 从站ID
AvgResponseTime time.Duration // 平均响应时间
MaxResponseTime time.Duration // 最大响应时间
ErrorRate float64 // 错误率
StabilityScore float64 // 稳定性得分 (0-1)
OptimalTimeout time.Duration // 最优超时时间
OptimalInterval time.Duration // 最优采集间隔
RetryCount int // 重试次数
BatchSize int // 批量读取大小
ProtocolParams map[string]interface{} // 协议特定参数
LastUpdated time.Time // 最后更新时间
采集成功率 float64 // 采集成功率
异常点位数量 int // 异常点位数量
连续失败次数 int // 连续失败次数
// RTT相关字段
RTTSamples []int64 // RTT采样数据
RTTSampleWindow int // RTT采样窗口大小
EWMARTT int64 // 指数加权移动平均RTT
// MTU相关字段
CurrentMTU int // 当前MTU值
MaxMTU int // 最大MTU值
MinMTU int // 最小MTU值
// Gap合并相关字段
CurrentGap int // 当前Gap值
MaxGap int // 最大Gap值
GapFillStrategy int // Gap填充策略
// 心跳相关字段
HeartbeatInterval int // 心跳间隔
LastActivity time.Time // 最后活跃时间
}
2.2.2 智能适配接口
// 智能适配接口
type DeviceAdapter interface {
// 获取设备画像
GetProfile(deviceID string) (*DeviceCommunicationProfile, error)
// 更新设备画像
UpdateProfile(deviceID string, responseTime time.Duration, err error, successCount, failCount int)
// 获取最优采集参数
GetOptimalParams(deviceID string) (map[string]interface{}, error)
// 标记设备状态
MarkDeviceStatus(deviceID string, status string)
// 获取设备状态
GetDeviceStatus(deviceID string) string
// 批量优化
OptimizeBatchCollection(devices []string) []string
// RTT统计
UpdateRTT(deviceID string, rtt int64)
// MTU协商
NegotiateMTU(deviceID string) error
// Gap优化
OptimizeGap(deviceID string) int
}
2.2.3 协议适配器接口
// 协议适配器接口
type ProtocolAdapter interface {
// 协议类型
ProtocolType() string
// 初始化设备参数
InitDeviceParams(deviceID string) error
// 调整协议特定参数
AdjustProtocolParams(profile *DeviceCommunicationProfile) error
// 验证设备响应
ValidateResponse(deviceID string, response interface{}, err error) bool
// 获取协议默认参数
GetDefaultParams() map[string]interface{}
// 获取协议MTU范围
GetMTURange() (min, max int)
// 计算最优Gap值
CalculateOptimalGap(profile *DeviceCommunicationProfile) int
// 构建心跳包
BuildHeartbeatPacket(deviceID string) ([]byte, error)
}
2.3 实现架构
2.3.1 核心组件
- DeviceAdapterManager：设备适配器管理器，负责管理所有设备的画像和适配策略
- ProtocolAdapterRegistry：协议适配器注册表，管理不同协议的适配器
- CollectionScheduler：采集调度器，根据设备画像优化采集顺序和策略
- BatchOptimizer：批量采集优化器，优化同一总线上的设备采集
- RTTManager：RTT管理器，负责RTT统计和分析
- MTUManager：MTU管理器，负责MTU动态协商
- GapOptimizer：Gap优化器，负责批量读取的Gap合并策略
2.3.2 数据流
- 数据采集流程：
- 采集前：获取设备最优参数，包括RTT、MTU、Gap等
- 采集中：记录响应时间、RTT、错误等信息
- 采集后：更新设备画像，包括RTT统计、MTU调整、Gap优化
- 参数调整流程：
- 基于历史RTT数据计算EWMA
- 根据RTT动态调整采集间隔和超时时间
- 基于MTU协商结果调整批量读取大小
- 根据RTT和MTU动态调整Gap值
2.4 多协议支持
2.4.1 协议差异处理
| 协议 | 特有参数 | 适配策略 |
|---|---|---|
| Modbus | 从站地址、功能码、寄存器范围 | 批量读取优化、帧间隔调整、RTT自适应、MTU协商 |
| TCP | 连接超时、重试次数 | 连接池管理、并发控制、RTT自适应 |
| BACnet | 对象ID、属性ID | 批量读属性、COV订阅、RTT自适应 |
| S7 | DB块、数据类型、偏移量 | 批量读取、数据块优化、MTU协商 |
| OPC UA | 节点ID、采样间隔 | 订阅优化、批处理、RTT自适应 |
| Serial | 波特率、奇偶校验、停止位 | 串口参数调整、缓冲区管理、RTT自适应 |
2.4.2 统一接口实现
2.5 影子设备设计
2.5.1 核心设计理念
- 统一内部数据面：Shadow Device 作为唯一可读写模型，所有数据操作通过 Shadow Core 进行。
- 统一控制面：所有写操作进入原子接口与审计链，确保数据一致性。
- 统一扩展面：南向/北向均可热插拔，系统重启零丢数。
2.5.2 分层架构
┌───────────────────────────────────────────────────────────────────────────┐
│ Northbound Adapters (MQTT / OPC UA / REST / Modbus-TCP Server / Others) │
│ - AuthN/AuthZ - Rate Limit - Audit Log - Dynamic Mount/Unmount │
└───────────────────────────────┬───────────────────────────────────────────┘
│
Shadow Service API
read/write/subscribe/consistency-check
│
┌───────────────────────────────┴───────────────────────────────────────────┐
│ Shadow Core (唯一内部模型中心) │
│ - Real Shadow Store (In-Memory) │
│ - Virtual Shadow Engine (公式依赖图 + 增量计算) │
│ - Edge Compute Engine (差值/步长/滑动平均/阈值) │
│ - Atomic Dispatcher (CAS + 序列号 + QoS ACK) │
└───────────────────────────────┬───────────────────────────────────────────┘
│
Shadow Ingress
│
┌───────────────────────────────┴───────────────────────────────────────────┐
│ Points Layer │
│ - Normalize - Unit Convert - Timestamp Align - Batch Publish QoS0/1/2 │
└───────────────────────────────┬───────────────────────────────────────────┘
│
Southbound Drivers (Modbus/CAN/IEC/...)
2.5.3 关键流程
- 采集入影子时序：
- 驱动采集原始值写入 Points。
- Points 完成归一化并以统一报文批量发布至 ShadowIngress。
- Shadow Core 原子落地内存。
- 触发边缘计算与虚拟公式增量更新。
- 北向适配器通过统一 API 拉取或订阅结果。
- 北向写入控制时序：
- 北向适配器请求授权与限流。
- 调用 Shadow Core 写入接口。
- Shadow Core 更新内存并通知订阅者。
- 记录审计日志并返回版本确认。
- 重启恢复时序：
- 进程启动后加载配置（
config.db）。 - ScanEngine 重新采集并填充 Shadow Core 内存快照。
- 重建虚拟引擎依赖缓存。
- 恢复北向适配器游标并继续服务。
- 进程启动后加载配置（
2.5.4 核心数据结构
2.5.4.1 点位标准报文（Points -> ShadowIngress）
messageId: "uuid"
qos: 0|1|2
deviceId: "physical-device-id"
timestamp: "2026-03-04T12:00:00.000Z"
points:
- pointId: "pack_voltage"
value: 742.3
unit: "V"
quality: "good"
samplePeriodMs: 1000
meta:
source: "modbus-driver"
sequence: 123456
2.5.4.2 真实影子设备结构
shadowDeviceId: "shadow-rack-01"
physicalDeviceId: "rack-01"
version: 202603040001
updatedAt: "2026-03-04T12:00:00.001Z"
points:
pack_voltage:
value: 742.3
unit: "V"
rw: "r"
samplePeriodMs: 1000
2.5.4.3 虚拟影子设备结构
virtualDeviceId: "virtual-energy-mix-01"
version: 202603040010
formulaPoints:
A: "device1.temp + device2.humidity"
dependencies:
- "device1.temp"
- "device2.humidity"
2.5.4.4 设备历史快照结构
ts: 1741084800
data:
pack_voltage: 742.3
pack_current: 125.0
2.5.5 一致性校验
- 接口输入：时间点
t或版本号区间。 - 校验对象：真实影子、虚拟影子、北向缓存。
- 判定条件：三端
value + version + timestamp + quality全量一致。 - 输出结果：
pass/fail、差异点位、差异来源、修复建议。
2.5.6 性能与容量目标
- 影子单机读写延迟：P99 ≤ 5ms。
- 虚拟公式计算：P99 ≤ 15ms。
- 虚拟点位传播延迟：≤ 100ms（单核 2GHz 基准）。
- 稳态吞吐：5 万点/秒持续 24h 零丢数。
2.6 边缘运维与设备替换（Gossip 分布式配置同步）
2.6.1 阶段目标
- 建立基于 Gossip 的本地网络配置一致性机制。
- 每个节点将本机与最近邻节点配置增量同步到本机 bbolt DB。
- 支持设备替换与离线缓存，网络恢复后自动补同步。
- 提供冲突检测与自动合并，冲突全过程审计可追溯。
2.6.2 同步数据模型
device_uuid：Base58 编码，默认基于 LAN 口 MAC 生成；支持手动指定。config_content：配置文件完整内容（文本或二进制编码）。sync_timestamp：最近一次同步时间戳（UTC，毫秒精度）。config_sha256：配置内容 SHA-256 哈希。source_node：来源节点 UUID。version_vector：可选向量时钟，用于高级冲突定位。
2.6.3 UUID 规则
- 默认生成：
uuid = Base58(SHA-256(lan_mac))[0:22]。 - 手工指定：写入本地设备标识配置，重启后生效且优先级高于默认生成。
- 替换场景：新设备可继承旧 UUID，确保邻居节点将其识别为同一逻辑节点。
2.6.4 架构组件
- Gossip 通讯模块：节点发现、成员心跳、增量反熵同步。
- bbolt 持久化层：桶设计包括
configs、sync_meta、offline_queue、conflict_logs。 - 冲突解决引擎：时间戳最新优先（Last-Write-Wins）策略，保留审计轨迹。
- 离线补同步队列：断网时写入队列，恢复后按时间顺序重放并去重确认。
- 运维工具与监控 API：UUID 管理、同步状态监控、冲突处理。
2.6.5 同步流程
- 本地配置变更后计算 SHA-256 并写入 bbolt。
- Gossip 广播摘要；邻居比对后按需请求缺失配置内容。
- 接收端执行冲突检测，按最新时间戳合并并记录审计日志。
- 合并结果写入
configs并更新时间戳，触发二次扩散。 - 离线节点恢复后，自动从邻居拉取缺失版本并补齐。
2.6.6 异常与恢复策略
- 网络分区：分区内局部一致，合区后执行冲突归并与去重扩散。
- 设备掉线：节点状态标记为
suspect/down，保留其最近有效配置快照。 - 配置回滚：支持指定
config_sha256或时间点回滚，并通过 Gossip 广播回滚版本。 - 时钟漂移：启用 NTP 校时；漂移超阈值时标记低可信并告警。
2.6.7 监控 API
GET /api/v1/ops/sync/nodes：节点在线状态与最近同步时间。GET /api/v1/ops/sync/status：全局同步进度、队列长度、补同步耗时。GET /api/v1/ops/sync/conflicts：冲突列表、处理结果、责任节点。POST /api/v1/ops/sync/reconcile：触发手动补同步与重算。POST /api/v1/ops/uuid/assign：手工设置设备 UUID（重启后生效）。
2.7 RTT管理器实现
// RTT管理器
type RTTManager struct {
rttData map[string][]int64
ewmaData map[string]int64
sampleWindow int
alpha float64
mu sync.RWMutex
}
func NewRTTManager() *RTTManager {
return &RTTManager{
rttData: make(map[string][]int64),
ewmaData: make(map[string]int64),
sampleWindow: 20,
alpha: 0.125,
}
}
func (r *RTTManager) UpdateRTT(deviceID string, rtt int64) {
r.mu.Lock()
defer r.mu.Unlock()
if _, exists := r.rttData[deviceID]; !exists {
r.rttData[deviceID] = make([]int64, 0, r.sampleWindow)
r.ewmaData[deviceID] = rtt
}
r.rttData[deviceID] = append(r.rttData[deviceID], rtt)
if len(r.rttData[deviceID]) > r.sampleWindow {
r.rttData[deviceID] = r.rttData[deviceID][1:]
}
// 计算EWMA
currentEWMA := r.ewmaData[deviceID]
newEWMA := int64(float64(currentEWMA)*(1-r.alpha) + float64(rtt)*r.alpha)
r.ewmaData[deviceID] = newEWMA
}
func (r *RTTManager) GetEWMARTT(deviceID string) int64 {
r.mu.RLock()
defer r.mu.RUnlock()
if ewma, exists := r.ewmaData[deviceID]; exists {
return ewma
}
return 0
}
2.8 MTU管理器实现
// MTU管理器
type MTUManager struct {
mtuData map[string]int
mtuHistory map[string][]MTUNegotiationRecord
minMTU map[string]int
maxMTU map[string]int
hysteresis float64
mu sync.RWMutex
}
func NewMTUManager() *MTUManager {
return &MTUManager{
mtuData: make(map[string]int),
mtuHistory: make(map[string][]MTUNegotiationRecord),
minMTU: make(map[string]int),
maxMTU: make(map[string]int),
hysteresis: 0.1,
}
}
func (m *MTUManager) NegotiateMTU(deviceID string, rtt int64) {
m.mu.Lock()
defer m.mu.Unlock()
if _, exists := m.mtuData[deviceID]; !exists {
m.mtuData[deviceID] = 1500
m.minMTU[deviceID] = 64
m.maxMTU[deviceID] = 4096
m.mtuHistory[deviceID] = make([]MTUNegotiationRecord, 0)
}
// 简单的MTU协商逻辑
currentMTU := m.mtuData[deviceID]
minMTU := m.minMTU[deviceID]
maxMTU := m.maxMTU[deviceID]
// 根据RTT调整MTU
if rtt < 10000 { // RTT < 10ms
// 尝试增加MTU
newMTU := currentMTU * 2
if newMTU > maxMTU {
newMTU = maxMTU
}
if newMTU > currentMTU {
m.mtuData[deviceID] = newMTU
m.mtuHistory[deviceID] = append(m.mtuHistory[deviceID], MTUNegotiationRecord{
AttemptValue: newMTU,
ResponseTime: rtt,
RetryCount: 0,
Success: true,
Timestamp: time.Now(),
})
}
} else if rtt > 50000 { // RTT > 50ms
// 尝试减少MTU
newMTU := currentMTU / 2
if newMTU < minMTU {
newMTU = minMTU
}
if newMTU < currentMTU {
m.mtuData[deviceID] = newMTU
m.mtuHistory[deviceID] = append(m.mtuHistory[deviceID], MTUNegotiationRecord{
AttemptValue: newMTU,
ResponseTime: rtt,
RetryCount: 0,
Success: true,
Timestamp: time.Now(),
})
}
}
}
func (m *MTUManager) GetCurrentMTU(deviceID string) int {
m.mu.RLock()
defer m.mu.RUnlock()
if mtu, exists := m.mtuData[deviceID]; exists {
return mtu
}
return 1500
}
2.9 Gap优化器实现
// Gap优化器
type GapOptimizer struct {
gapData map[string]int
mu sync.RWMutex
}
func NewGapOptimizer() *GapOptimizer {
return &GapOptimizer{
gapData: make(map[string]int),
}
}
func (g *GapOptimizer) OptimizeGap(deviceID string, mtu int, rtt int64) int {
g.mu.Lock()
defer g.mu.Unlock()
// 根据MTU和RTT计算最优Gap值
// MTU越大，允许的Gap越大
// RTT越小，允许的Gap越大
baseGap := mtu / 100
if baseGap < 0 {
baseGap = 0
}
if baseGap > 20 {
baseGap = 20
}
// 根据RTT调整
rttFactor := 1.0
if rtt > 20000 {
rttFactor = 0.5
} else if rtt > 10000 {
rttFactor = 0.8
}
optimalGap := int(float64(baseGap) * rttFactor)
if optimalGap < 0 {
optimalGap = 0
}
g.gapData[deviceID] = optimalGap
return optimalGap
}
2.10 并发采集优化
2.10.1 智能调度算法
// 采集调度器
type CollectionScheduler struct {
adapterManager *DeviceAdapterManager
}
// 优化采集顺序
func (s *CollectionScheduler) OptimizeCollectionOrder(devices []string) []string {
// 获取所有设备的画像
profiles := make(map[string]*DeviceCommunicationProfile)
for _, deviceID := range devices {
if profile, err := s.adapterManager.GetProfile(deviceID); err == nil {
profiles[deviceID] = profile
}
}
// 根据稳定性得分、响应时间和RTT排序
sort.Slice(devices, func(i, j int) bool {
profileI, existsI := profiles[devices[i]]
profileJ, existsJ := profiles[devices[j]]
// 优先采集稳定设备
if existsI && existsJ {
if profileI.StabilityScore != profileJ.StabilityScore {
return profileI.StabilityScore > profileJ.StabilityScore
}
// 稳定性相同时，优先采集响应快的设备
if profileI.AvgResponseTime != profileJ.AvgResponseTime {
return profileI.AvgResponseTime < profileJ.AvgResponseTime
}
// 响应时间相同时，优先采集RTT小的设备
return profileI.EWMARTT < profileJ.EWMARTT
}
// 没有画像的设备放在最后
return existsI && !existsJ
})
return devices
}
// 批量采集优化
func (s *CollectionScheduler) OptimizeBatchCollection(devices []string) [][]string {
// 根据协议类型和总线分组
protocolGroups := make(map[string][]string)
for _, deviceID := range devices {
if profile, err := s.adapterManager.GetProfile(deviceID); err == nil {
key := profile.ProtocolType
// 对于Modbus等协议，按从站地址进一步分组
if strings.HasPrefix(profile.ProtocolType, "modbus") {
if slaveID, ok := profile.SlaveID.(string); ok {
key += ":" + slaveID
} else if slaveID, ok := profile.SlaveID.(uint8); ok {
key += ":" + strconv.Itoa(int(slaveID))
}
}
protocolGroups[key] = append(protocolGroups[key], deviceID)
}
}
// 为每个协议组创建批处理
batches := make([][]string, 0)
for _, group := range protocolGroups {
// 根据设备特性进一步分组
stableDevices := make([]string, 0)
unstableDevices := make([]string, 0)
for _, deviceID := range group {
if profile, err := s.adapterManager.GetProfile(deviceID); err == nil {
if profile.StabilityScore >= 0.8 {
stableDevices = append(stableDevices, deviceID)
} else {
unstableDevices = append(unstableDevices, deviceID)
}
}
}
// 稳定设备可以批量采集
if len(stableDevices) > 0 {
batches = append(batches, stableDevices)
}
// 不稳定设备单独采集
for _, deviceID := range unstableDevices {
batches = append(batches, []string{deviceID})
}
}
return batches
}
2.11 与现有系统集成
2.11.1 集成点
- ChannelManager集成：在设备采集循环中集成智能适配逻辑
- 驱动层集成：在驱动读取操作中应用最优参数，包括RTT、MTU、Gap等
- 状态管理集成：将设备画像与现有状态管理系统结合
- Diagnostics 集成：设备画像摘要暴露于现有 diagnostics API，供 UI 轮询与 curl 巡检
- 心跳机制集成：实现智能业务心跳，减少不必要的心跳包
2.11.2 集成实现
// 在ChannelManager的deviceLoop中集成智能适配
func (cm *ChannelManager) deviceLoop(dev *model.Device, d drv.Driver, ch *model.Channel) {
// 初始化设备适配器
adapter := globalDeviceAdapter
ticker := time.NewTicker(time.Duration(dev.Interval))
defer ticker.Stop()
node := cm.stateManager.GetNode(dev.ID)
if node == nil {
zap.L().Warn("Device node not found in state manager", zap.String("device", dev.Name))
return
}
var offset time.Duration
for i := range ch.Devices {
if ch.Devices[i].ID == dev.ID {
offset = time.Duration(i) * 12 * time.Millisecond
break
}
}
for {
select {
case <-cm.ctx.Done():
return
case <-dev.StopChan:
return
case <-ticker.C:
time.Sleep(offset)
if !cm.stateManager.ShouldCollect(node) {
zap.L().Debug("Device skipped collection",
zap.String("device", dev.Name),
zap.Any("state", node.Runtime.State),
zap.Time("next_retry", node.Runtime.NextRetryTime))
continue
}
cm.collectDeviceWithAdapter(dev, d, ch, node, adapter)
}
}
}
// 带智能适配的设备采集
func (cm *ChannelManager) collectDeviceWithAdapter(dev *model.Device, d drv.Driver, ch *model.Channel, node *DeviceNodeTemplate, adapter DeviceAdapter) {
start := time.Now()
defer func() {
duration := time.Since(start)
zap.L().Info("Device collection cycle finished",
zap.String("device", dev.Name),
zap.Int("point_count", len(dev.Points)),
zap.String("duration", fmt.Sprintf("%.3fs", duration.Seconds())),
zap.Int64("rtt_us", duration.Microseconds()),
)
}()
zap.L().Info("PollStart", zap.String("device", dev.Name), zap.Time("ts", time.Now()))
// 获取最优参数
params, err := adapter.GetOptimalParams(dev.ID)
if err != nil {
zap.L().Warn("Failed to get optimal params", zap.String("device", dev.Name), zap.Error(err))
}
// 应用最优超时时间
timeout := 5 * time.Second
if params != nil {
if timeoutVal, ok := params["timeout"].(time.Duration); ok {
timeout = timeoutVal
}
}
if node.Runtime.State != NodeStateOnline {
timeout = 200 * time.Millisecond
}
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
// 获取驱动互斥锁
cm.mu.RLock()
mu, okMu := cm.driverMus[ch.ID]
cm.mu.RUnlock()
if okMu {
mu.Lock()
defer mu.Unlock()
}
// 设置从机 ID
if slaveID, ok := dev.Config["slave_id"]; ok {
if slaveIDUint, ok := slaveID.(float64); ok {
d.SetSlaveID(uint8(slaveIDUint))
} else if slaveIDInt, ok := slaveID.(int); ok {
d.SetSlaveID(uint8(slaveIDInt))
}
}
// 设置设备配置
config := make(map[string]any)
for k, v := range dev.Config {
config[k] = v
}
config["_internal_device_id"] = dev.ID
// 应用MTU和Gap参数
if params != nil {
if mtu, ok := params["mtu"].(int); ok {
config["mtu"] = mtu
}
if gap, ok := params["gap"].(int); ok {
config["gap"] = gap
}
}
if err := d.SetDeviceConfig(config); err != nil {
zap.L().Error("Failed to set device config", zap.String("device", dev.Name), zap.Error(err))
}
// Ensure DeviceID is set on points for the driver
for i := range dev.Points {
dev.Points[i].DeviceID = dev.ID
}
// 读取点位数据
results, err := d.ReadPoints(ctx, dev.Points)
responseTime := time.Since(start)
// 统计采集结果
successCount := 0
failCount := 0
now := time.Now()
for _, result := range results {
// 发送到管道
val := model.Value{
ChannelID: ch.ID,
DeviceID: dev.ID,
PointID: result.PointID,
Value: result.Value,
Quality: result.Quality,
TS: now,
}
// 推入数据管道，驱动存储与WebSocket广播
cm.pipeline.Push(val)
// 统计成功/失败
if result.Quality == "Good" {
successCount++
} else {
failCount++
}
}
// 使用 FinalizeCollect 进行状态裁决
if len(dev.Points) > 0 && len(results) == 0 {
failCount = len(dev.Points) // 假设所有点位都失败
}
// 更新设备画像
adapter.UpdateProfile(dev.ID, responseTime, err, successCount, failCount)
// Update Device Metrics
if mc := model.GetGlobalMetricsCollector(); mc != nil {
mc.UpdateDeviceMetrics(dev.ID, func(m *model.DeviceMetrics) {
total := successCount + failCount
if total > 0 {
m.PointSuccessRate = float64(successCount) / float64(total)
} else {
m.PointSuccessRate = 0
}
m.AbnormalPoints = failCount
if failCount == len(dev.Points) && len(dev.Points) > 0 {
m.ConsecutiveFailures++
} else {
m.ConsecutiveFailures = 0
}
m.LastCollectTime = now
m.State = int(node.Runtime.State)
})
}
collectCtx := &CollectContext{
TotalCmd: successCount + failCount,
SuccessCmd: successCount,
FailCmd: failCount,
}
cm.stateManager.FinalizeCollect(node, collectCtx)
}
2.12 配置与监控
2.12.1 配置结构
2.13 通用点位降级框架
2.13.1 设计目标
- 协议无关性：为所有支持的协议提供统一的点位降级机制
- 智能降级：基于点位失败次数和时间的智能降级策略
- 动态恢复：当点位恢复正常时自动重新纳入采集
- 指数级退避：对降级点位采用指数级间隔探测机制
- 与设备画像集成：将点位降级状态与设备画像系统集成
- 可扩展性：易于添加新协议的点位降级支持
2.13.2 核心数据结构
// 点位状态管理结构
type PointStatus struct {
PointID string // 点位唯一标识
DeviceID string // 所属设备ID
ProtocolType string // 协议类型
FailureCount int // 连续失败次数
LastSuccessTime time.Time // 最后成功时间
LastFailureTime time.Time // 最后失败时间
Degraded bool // 是否处于降级状态
LastProbeTime time.Time // 最后探测时间
ProbeInterval int // 当前探测间隔（秒）
MaxProbeInterval int // 最大探测间隔（秒）
Status string // 点位状态：normal, warning, degraded, recovering
Metadata map[string]interface{} // 协议特定元数据
}
// 点位降级管理器
type PointDegradationManager struct {
pointStatuses map[string]*PointStatus // 点位状态映射
protocolAdapters map[string]ProtocolPointAdapter // 协议点位适配器
failureThreshold int // 失败次数阈值
maxProbeInterval int // 最大探测间隔
mu sync.RWMutex // 读写锁
}
// 协议点位适配器接口
type ProtocolPointAdapter interface {
// 协议类型
ProtocolType() string
// 初始化点位状态
InitPointStatus(pointID, deviceID string) *PointStatus
// 判定点位是否失败
IsPointFailure(pointID string, response interface{}, err error) bool
// 处理点位成功
HandlePointSuccess(status *PointStatus)
// 处理点位失败
HandlePointFailure(status *PointStatus)
// 检查是否需要探测
ShouldProbe(status *PointStatus) bool
// 执行点位探测
ProbePoint(pointID, deviceID string) bool
// 获取协议特定的失败阈值
GetFailureThreshold() int
}
2.13.3 核心实现
// 创建点位降级管理器
func NewPointDegradationManager() *PointDegradationManager {
return &PointDegradationManager{
pointStatuses: make(map[string]*PointStatus),
protocolAdapters: make(map[string]ProtocolPointAdapter),
failureThreshold: 3,
maxProbeInterval: 1024,
}
}
// 注册协议点位适配器
func (p *PointDegradationManager) RegisterProtocolAdapter(adapter ProtocolPointAdapter) {
p.mu.Lock()
defer p.mu.Unlock()
p.protocolAdapters[adapter.ProtocolType()] = adapter
}
// 获取点位状态
func (p *PointDegradationManager) GetPointStatus(pointID, deviceID, protocolType string) *PointStatus {
p.mu.RLock()
key := fmt.Sprintf("%s:%s", deviceID, pointID)
status, exists := p.pointStatuses[key]
p.mu.RUnlock()
if !exists {
p.mu.Lock()
defer p.mu.Unlock()
// 双重检查
if status, exists = p.pointStatuses[key]; !exists {
adapter, adapterExists := p.protocolAdapters[protocolType]
if adapterExists {
status = adapter.InitPointStatus(pointID, deviceID)
} else {
status = &PointStatus{
PointID: pointID,
DeviceID: deviceID,
ProtocolType: protocolType,
FailureCount: 0,
LastSuccessTime: time.Time{},
LastFailureTime: time.Time{},
Degraded: false,
LastProbeTime: time.Time{},
ProbeInterval: 8,
MaxProbeInterval: p.maxProbeInterval,
Status: "normal",
Metadata: make(map[string]interface{}),
}
}
p.pointStatuses[key] = status
}
}
return status
}
// 更新点位状态
func (p *PointDegradationManager) UpdatePointStatus(pointID, deviceID, protocolType string, success bool, response interface{}, err error) {
key := fmt.Sprintf("%s:%s", deviceID, pointID)
status := p.GetPointStatus(pointID, deviceID, protocolType)
p.mu.Lock()
defer p.mu.Unlock()
adapter, exists := p.protocolAdapters[protocolType]
if exists {
if success && !adapter.IsPointFailure(pointID, response, err) {
adapter.HandlePointSuccess(status)
} else {
adapter.HandlePointFailure(status)
}
} else {
if success && err == nil {
// 通用成功处理
status.FailureCount = 0
status.LastSuccessTime = time.Now()
status.Degraded = false
status.Status = "normal"
status.ProbeInterval = 8
} else {
// 通用失败处理
status.FailureCount++
status.LastFailureTime = time.Now()
if status.FailureCount >= p.failureThreshold {
status.Degraded = true
status.Status = "degraded"
} else if status.FailureCount > 0 {
status.Status = "warning"
}
}
}
}
// 检查点位是否应该被采集
func (p *PointDegradationManager) ShouldCollectPoint(pointID, deviceID, protocolType string) bool {
status := p.GetPointStatus(pointID, deviceID, protocolType)
if !status.Degraded {
return true
}
// 检查是否需要探测
adapter, exists := p.protocolAdapters[protocolType]
if exists {
return adapter.ShouldProbe(status)
}
// 通用探测逻辑
now := time.Now()
if now.Sub(status.LastProbeTime).Seconds() >= float64(status.ProbeInterval) {
status.LastProbeTime = now
// 指数级增加探测间隔
newInterval := status.ProbeInterval * 2
if newInterval > status.MaxProbeInterval {
newInterval = status.MaxProbeInterval
}
status.ProbeInterval = newInterval
return true
}
return false
}
// 批量检查点位采集状态
func (p *PointDegradationManager) FilterCollectiblePoints(points []model.Point, deviceID, protocolType string) []model.Point {
var collectiblePoints []model.Point
for _, point := range points {
if p.ShouldCollectPoint(point.ID, deviceID, protocolType) {
collectiblePoints = append(collectiblePoints, point)
}
}
return collectiblePoints
}
2.13.4 协议特定实现
2.13.4.1 Modbus点位适配器
// Modbus点位适配器
type ModbusPointAdapter struct {
failureThreshold int
}
func NewModbusPointAdapter() *ModbusPointAdapter {
return &ModbusPointAdapter{
failureThreshold: 3,
}
}
func (m *ModbusPointAdapter) ProtocolType() string {
return "modbus"
}
func (m *ModbusPointAdapter) InitPointStatus(pointID, deviceID string) *PointStatus {
return &PointStatus{
PointID: pointID,
DeviceID: deviceID,
ProtocolType: "modbus",
FailureCount: 0,
LastSuccessTime: time.Time{},
LastFailureTime: time.Time{},
Degraded: false,
LastProbeTime: time.Time{},
ProbeInterval: 8,
MaxProbeInterval: 1024,
Status: "normal",
Metadata: make(map[string]interface{}),
}
}
func (m *ModbusPointAdapter) IsPointFailure(pointID string, response interface{}, err error) bool {
if err != nil {
return true
}
// 检查响应数据是否有效
// ...
return false
}
func (m *ModbusPointAdapter) HandlePointSuccess(status *PointStatus) {
status.FailureCount = 0
status.LastSuccessTime = time.Now()
status.Degraded = false
status.Status = "normal"
status.ProbeInterval = 8
}
func (m *ModbusPointAdapter) HandlePointFailure(status *PointStatus) {
status.FailureCount++
status.LastFailureTime = time.Now()
if status.FailureCount >= m.failureThreshold {
status.Degraded = true
status.Status = "degraded"
} else if status.FailureCount > 0 {
status.Status = "warning"
}
}
func (m *ModbusPointAdapter) ShouldProbe(status *PointStatus) bool {
now := time.Now()
if now.Sub(status.LastProbeTime).Seconds() >= float64(status.ProbeInterval) {
status.LastProbeTime = now
newInterval := status.ProbeInterval * 2
if newInterval > status.MaxProbeInterval {
newInterval = status.MaxProbeInterval
}
status.ProbeInterval = newInterval
return true
}
return false
}
func (m *ModbusPointAdapter) ProbePoint(pointID, deviceID string) bool {
// 执行Modbus点位探测
// ...
return false
}
func (m *ModbusPointAdapter) GetFailureThreshold() int {
return m.failureThreshold
}
2.13.4.2 BACnet点位适配器
// BACnet点位适配器
type BACnetPointAdapter struct {
failureThreshold int
}
func NewBACnetPointAdapter() *BACnetPointAdapter {
return &BACnetPointAdapter{
failureThreshold: 3,
}
}
func (b *BACnetPointAdapter) ProtocolType() string {
return "bacnet"
}
// 其他方法实现...
2.13.4.3 OPC UA点位适配器
// OPC UA点位适配器
type OPCUAPointAdapter struct {
failureThreshold int
}
func NewOPCUAPointAdapter() *OPCUAPointAdapter {
return &OPCUAPointAdapter{
failureThreshold: 3,
}
}
func (o *OPCUAPointAdapter) ProtocolType() string {
return "opcua"
}
// 其他方法实现...
2.13.4.4 S7点位适配器
// S7点位适配器
type S7PointAdapter struct {
failureThreshold int
}
func NewS7PointAdapter() *S7PointAdapter {
return &S7PointAdapter{
failureThreshold: 3,
}
}
func (s *S7PointAdapter) ProtocolType() string {
return "s7"
}
// 其他方法实现...
2.13.5 与现有系统集成
2.13.5.1 与ChannelManager集成
基于通用点位降级策略的设计，我们优化了接入点，将点位降级逻辑集成到协议驱动的批次构建过程中，并通过抽象层设计确保方案适用于所有协议设备，这样可以更贴近各协议特性，提高降级策略的有效性和通用性。
// 扩展DeviceNodeTemplate结构，添加点位状态管理字段
type DeviceNodeTemplate struct {
// 现有字段...
PointFailureMap map[string]int `json:"-"` // 点位失败次数映射
PointLastSuccessTime map[string]time.Time `json:"-"` // 点位最后成功时间
PointDegradedMap map[string]bool `json:"-"` // 点位降级状态映射
PointLastProbeTime map[string]time.Time `json:"-"` // 点位最后探测时间
PointProbeInterval map[string]int `json:"-"` // 点位当前探测间隔（秒）
// 其他字段...
}
// 在NewDeviceNode方法中初始化这些字段
func NewDeviceNode() *DeviceNodeTemplate {
return &DeviceNodeTemplate{
// 其他初始化...
PointFailureMap: make(map[string]int),
PointLastSuccessTime: make(map[string]time.Time),
PointDegradedMap: make(map[string]bool),
PointLastProbeTime: make(map[string]time.Time),
PointProbeInterval: make(map[string]int),
// 其他初始化...
}
}
// 协议点位适配器接口，为不同协议提供点位降级逻辑
type ProtocolPointAdapter interface {
// 协议类型
ProtocolType() string
// 检查点位是否失败
IsPointFailure(pointID string, response interface{}, err error) bool
// 处理点位成功
HandlePointSuccess(node *DeviceNodeTemplate, pointID string)
// 处理点位失败
HandlePointFailure(node *DeviceNodeTemplate, pointID string)
// 检查点位是否应该被降级
ShouldDegradePoint(node *DeviceNodeTemplate, pointID string) bool
// 检查点位是否应该被探测
ShouldProbePoint(node *DeviceNodeTemplate, pointID string) bool
// 更新探测间隔
UpdateProbeInterval(node *DeviceNodeTemplate, pointID string)
// 构建协议特定的批次
BuildBatches(points []model.Point) [][]model.Point
}
// 协议适配器注册表
type ProtocolAdapterRegistry struct {
adapters map[string]ProtocolPointAdapter
}
// 注册协议适配器
func (r *ProtocolAdapterRegistry) RegisterAdapter(adapter ProtocolPointAdapter) {
r.adapters[adapter.ProtocolType()] = adapter
}
// 获取协议适配器
func (r *ProtocolAdapterRegistry) GetAdapter(protocolType string) ProtocolPointAdapter {
if adapter, ok := r.adapters[protocolType]; ok {
return adapter
}
// 返回默认适配器
return &DefaultPointAdapter{}
}
// 默认点位适配器
type DefaultPointAdapter struct {}
func (d *DefaultPointAdapter) ProtocolType() string {
return "default"
}
func (d *DefaultPointAdapter) IsPointFailure(pointID string, response interface{}, err error) bool {
return err != nil
}
func (d *DefaultPointAdapter) HandlePointSuccess(node *DeviceNodeTemplate, pointID string) {
resetPointFailure(node, pointID)
}
func (d *DefaultPointAdapter) HandlePointFailure(node *DeviceNodeTemplate, pointID string) {
incrementPointFailure(node, pointID)
}
func (d *DefaultPointAdapter) ShouldDegradePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldDegradePoint(pointID, node)
}
func (d *DefaultPointAdapter) ShouldProbePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldProbePoint(pointID, node)
}
func (d *DefaultPointAdapter) UpdateProbeInterval(node *DeviceNodeTemplate, pointID string) {
updateProbeInterval(pointID, node)
}
func (d *DefaultPointAdapter) BuildBatches(points []model.Point) [][]model.Point {
// 默认实现：每个点位一个批次
var batches [][]model.Point
for _, point := range points {
batches = append(batches, []model.Point{point})
}
return batches
}
// Modbus点位适配器
type ModbusPointAdapter struct {
batchSize int
}
func NewModbusPointAdapter() *ModbusPointAdapter {
return &ModbusPointAdapter{batchSize: 100}
}
func (m *ModbusPointAdapter) ProtocolType() string {
return "modbus"
}
func (m *ModbusPointAdapter) IsPointFailure(pointID string, response interface{}, err error) bool {
if err != nil {
return true
}
// Modbus特定的失败检查
// ...
return false
}
func (m *ModbusPointAdapter) HandlePointSuccess(node *DeviceNodeTemplate, pointID string) {
resetPointFailure(node, pointID)
}
func (m *ModbusPointAdapter) HandlePointFailure(node *DeviceNodeTemplate, pointID string) {
incrementPointFailure(node, pointID)
}
func (m *ModbusPointAdapter) ShouldDegradePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldDegradePoint(pointID, node)
}
func (m *ModbusPointAdapter) ShouldProbePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldProbePoint(pointID, node)
}
func (m *ModbusPointAdapter) UpdateProbeInterval(node *DeviceNodeTemplate, pointID string) {
updateProbeInterval(pointID, node)
}
func (m *ModbusPointAdapter) BuildBatches(points []model.Point) [][]model.Point {
// Modbus特定的批次构建逻辑
var batches [][]model.Point
var currentBatch []model.Point
for _, point := range points {
if len(currentBatch) >= m.batchSize {
batches = append(batches, currentBatch)
currentBatch = []model.Point{point}
} else {
currentBatch = append(currentBatch, point)
}
}
if len(currentBatch) > 0 {
batches = append(batches, currentBatch)
}
return batches
}
// 在协议驱动中添加批次构建函数，集成点位降级逻辑
func buildProtocolBatchesWithDegradation(protocolType string, points []model.Point, node *DeviceNodeTemplate, registry *ProtocolAdapterRegistry) [][]model.Point {
var filtered []model.Point
adapter := registry.GetAdapter(protocolType)
for _, point := range points {
// 检查点位是否应该被降级
if adapter.ShouldDegradePoint(node, point.ID) {
// 检查是否达到探测间隔
if adapter.ShouldProbePoint(node, point.ID) {
// 允许探测
filtered = append(filtered, point)
// 更新探测时间和间隔
adapter.UpdateProbeInterval(node, point.ID)
} else {
// 跳过降级点位
zap.L().Debugf("[点位降级] 跳过点位 %s，失败次数: %d", point.ID, node.PointFailureMap[point.ID])
}
} else {
// 正常点位，加入采集列表
filtered = append(filtered, point)
}
}
// 使用协议适配器构建批次
return adapter.BuildBatches(filtered)
}
// 检查点位是否应该被降级
func shouldDegradePoint(pointID string, node *DeviceNodeTemplate) bool {
if node == nil || node.PointFailureMap == nil {
return false
}
if failCount, ok := node.PointFailureMap[pointID]; ok && failCount > 3 {
return true
}
return false
}
// 检查点位是否应该被探测
func shouldProbePoint(pointID string, node *DeviceNodeTemplate) bool {
if node == nil || node.PointLastProbeTime == nil || node.PointProbeInterval == nil {
return true
}
lastProbe, ok := node.PointLastProbeTime[pointID]
if !ok {
return true
}
interval, ok := node.PointProbeInterval[pointID]
if !ok {
interval = 8 // 默认8秒
}
return time.Since(lastProbe).Seconds() >= float64(interval)
}
// 更新探测间隔（指数级退避）
func updateProbeInterval(pointID string, node *DeviceNodeTemplate) {
if node == nil || node.PointProbeInterval == nil || node.PointLastProbeTime == nil {
return
}
interval, ok := node.PointProbeInterval[pointID]
if !ok {
interval = 8 // 默认8秒
}
newInterval := interval * 2
if newInterval > 1024 { // 最大1024秒
newInterval = 1024
}
node.PointProbeInterval[pointID] = newInterval
node.PointLastProbeTime[pointID] = time.Now()
}
// 在ChannelManager的collectDevice中集成点位降级
func (cm *ChannelManager) collectDevice(dev *model.Device, d drv.Driver, ch *model.Channel) {
start := time.Now()
defer func() {
duration := time.Since(start)
zap.L().Info("Device collection cycle finished",
zap.String("device", dev.Name),
zap.Int("point_count", len(dev.Points)),
zap.String("duration", fmt.Sprintf("%.3fs", duration.Seconds())),
zap.Int64("rtt_us", duration.Microseconds()),
)
}()
zap.L().Info("PollStart", zap.String("device", dev.Name), zap.Time("ts", time.Now()))
node := cm.stateManager.GetNode(dev.ID)
if node == nil {
zap.L().Warn("Device node not found in state manager", zap.String("device", dev.Name))
return
}
if !cm.stateManager.ShouldCollect(node) {
zap.L().Debug("Device skipped collection",
zap.String("device", dev.Name),
zap.Any("state", node.Runtime.State),
zap.Time("next_retry", node.Runtime.NextRetryTime))
return
}
// 获取协议类型
protocolType := "modbus" // 默认协议，实际应从设备配置中获取
if proto, ok := dev.Config["protocol"]; ok {
if protoStr, ok := proto.(string); ok {
protocolType = protoStr
}
}
// 构建带点位降级的采集批次
// 假设已经初始化了协议适配器注册表
registry := getProtocolAdapterRegistry()
batches := buildProtocolBatchesWithDegradation(protocolType, dev.Points, node, registry)
if len(batches) == 0 {
zap.L().Warn("No collectible points", zap.String("device", dev.Name))
return
}
// 应用最优超时时间
timeout := 5 * time.Second
ctx, cancel := context.WithTimeout(context.Background(), timeout)
defer cancel()
// 获取驱动互斥锁
cm.mu.RLock()
mu, okMu := cm.driverMus[ch.ID]
cm.mu.RUnlock()
if okMu {
mu.Lock()
defer mu.Unlock()
}
// 设置从机 ID
if slaveID, ok := dev.Config["slave_id"]; ok {
if slaveIDUint, ok := slaveID.(float64); ok {
d.SetSlaveID(uint8(slaveIDUint))
} else if slaveIDInt, ok := slaveID.(int); ok {
d.SetSlaveID(uint8(slaveIDInt))
}
}
// 设置设备配置
config := make(map[string]any)
for k, v := range dev.Config {
config[k] = v
}
config["_internal_device_id"] = dev.ID
if err := d.SetDeviceConfig(config); err != nil {
zap.L().Error("Failed to set device config", zap.String("device", dev.Name), zap.Error(err))
}
// 处理每个批次
successCount := 0
failCount := 0
now := time.Now()
for _, batch := range batches {
// 确保点位的DeviceID设置正确
for i := range batch {
batch[i].DeviceID = dev.ID
}
// 读取点位数据
results, err := d.ReadPoints(ctx, batch)
// 处理读取结果
adapter := registry.GetAdapter(protocolType)
for _, result := range results {
// 发送到管道
val := model.Value{
ChannelID: ch.ID,
DeviceID: dev.ID,
PointID: result.PointID,
Value: result.Value,
Quality: result.Quality,
TS: now,
}
// 推入数据管道，驱动存储与WebSocket广播
cm.pipeline.Push(val)
// 统计成功/失败并更新点位状态
if result.Quality == "Good" {
successCount++
// 重置失败次数和降级状态
adapter.HandlePointSuccess(node, result.PointID)
} else {
failCount++
// 增加失败次数，可能触发降级
adapter.HandlePointFailure(node, result.PointID)
}
}
// 处理未返回结果的点位
collectedPointIDs := make(map[string]bool)
for _, result := range results {
collectedPointIDs[result.PointID] = true
}
for _, point := range batch {
if !collectedPointIDs[point.ID] {
failCount++
// 增加失败次数，可能触发降级
adapter.HandlePointFailure(node, point.ID)
}
}
}
// 使用 FinalizeCollect 进行状态裁决
if len(dev.Points) > 0 && successCount+failCount == 0 {
failCount = len(dev.Points) // 假设所有点位都失败
}
// Update Device Metrics
if mc := model.GetGlobalMetricsCollector(); mc != nil {
mc.UpdateDeviceMetrics(dev.ID, func(m *model.DeviceMetrics) {
total := successCount + failCount
if total > 0 {
m.PointSuccessRate = float64(successCount) / float64(total)
} else {
m.PointSuccessRate = 0
}
m.AbnormalPoints = failCount
if failCount == len(dev.Points) && len(dev.Points) > 0 {
m.ConsecutiveFailures++
} else {
m.ConsecutiveFailures = 0
}
m.LastCollectTime = now
m.State = int(node.Runtime.State)
})
}
collectCtx := &CollectContext{
TotalCmd: successCount + failCount,
SuccessCmd: successCount,
FailCmd: failCount,
}
cm.stateManager.FinalizeCollect(node, collectCtx)
}
// 重置点位失败状态
func resetPointFailure(node *DeviceNodeTemplate, pointID string) {
if node == nil {
return
}
if node.PointFailureMap != nil {
node.PointFailureMap[pointID] = 0
}
if node.PointLastSuccessTime != nil {
node.PointLastSuccessTime[pointID] = time.Now()
}
if node.PointDegradedMap != nil {
node.PointDegradedMap[pointID] = false
}
if node.PointProbeInterval != nil {
node.PointProbeInterval[pointID] = 8 // 重置为默认值
}
}
// 增加点位失败次数
func incrementPointFailure(node *DeviceNodeTemplate, pointID string) {
if node == nil {
return
}
if node.PointFailureMap == nil {
node.PointFailureMap = make(map[string]int)
}
node.PointFailureMap[pointID]++
if node.PointLastSuccessTime != nil {
// 可以记录失败时间
}
// 检查是否需要降级
if node.PointFailureMap[pointID] > 3 {
if node.PointDegradedMap == nil {
node.PointDegradedMap = make(map[string]bool)
}
node.PointDegradedMap[pointID] = true
// 初始化探测间隔
if node.PointProbeInterval == nil {
node.PointProbeInterval = make(map[string]int)
}
if _, ok := node.PointProbeInterval[pointID]; !ok {
node.PointProbeInterval[pointID] = 8
}
// 初始化探测时间
if node.PointLastProbeTime == nil {
node.PointLastProbeTime = make(map[string]time.Time)
}
if _, ok := node.PointLastProbeTime[pointID]; !ok {
node.PointLastProbeTime[pointID] = time.Now()
}
}
}
// 全局协议适配器注册表
var globalProtocolAdapterRegistry *ProtocolAdapterRegistry
// 初始化协议适配器注册表
func initProtocolAdapterRegistry() {
globalProtocolAdapterRegistry = &ProtocolAdapterRegistry{
adapters: make(map[string]ProtocolPointAdapter),
}
// 注册默认适配器
globalProtocolAdapterRegistry.RegisterAdapter(&DefaultPointAdapter{})
// 注册Modbus适配器
globalProtocolAdapterRegistry.RegisterAdapter(NewModbusPointAdapter())
// 注册BACnet适配器
globalProtocolAdapterRegistry.RegisterAdapter(NewBACnetPointAdapter())
// 注册OPC UA适配器
globalProtocolAdapterRegistry.RegisterAdapter(NewOPCUAPointAdapter())
// 注册S7适配器
globalProtocolAdapterRegistry.RegisterAdapter(NewS7PointAdapter())
}
// 获取协议适配器注册表
func getProtocolAdapterRegistry() *ProtocolAdapterRegistry {
if globalProtocolAdapterRegistry == nil {
initProtocolAdapterRegistry()
}
return globalProtocolAdapterRegistry
}
// BACnet点位适配器
type BACnetPointAdapter struct {
batchSize int
}
func NewBACnetPointAdapter() *BACnetPointAdapter {
return &BACnetPointAdapter{batchSize: 50}
}
func (b *BACnetPointAdapter) ProtocolType() string {
return "bacnet"
}
func (b *BACnetPointAdapter) IsPointFailure(pointID string, response interface{}, err error) bool {
if err != nil {
return true
}
// BACnet特定的失败检查
// ...
return false
}
func (b *BACnetPointAdapter) HandlePointSuccess(node *DeviceNodeTemplate, pointID string) {
resetPointFailure(node, pointID)
}
func (b *BACnetPointAdapter) HandlePointFailure(node *DeviceNodeTemplate, pointID string) {
incrementPointFailure(node, pointID)
}
func (b *BACnetPointAdapter) ShouldDegradePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldDegradePoint(pointID, node)
}
func (b *BACnetPointAdapter) ShouldProbePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldProbePoint(pointID, node)
}
func (b *BACnetPointAdapter) UpdateProbeInterval(node *DeviceNodeTemplate, pointID string) {
updateProbeInterval(pointID, node)
}
func (b *BACnetPointAdapter) BuildBatches(points []model.Point) [][]model.Point {
// BACnet特定的批次构建逻辑
var batches [][]model.Point
var currentBatch []model.Point
for _, point := range points {
if len(currentBatch) >= b.batchSize {
batches = append(batches, currentBatch)
currentBatch = []model.Point{point}
} else {
currentBatch = append(currentBatch, point)
}
}
if len(currentBatch) > 0 {
batches = append(batches, currentBatch)
}
return batches
}
// OPC UA点位适配器
type OPCUAPointAdapter struct {
batchSize int
}
func NewOPCUAPointAdapter() *OPCUAPointAdapter {
return &OPCUAPointAdapter{batchSize: 200}
}
func (o *OPCUAPointAdapter) ProtocolType() string {
return "opcua"
}
func (o *OPCUAPointAdapter) IsPointFailure(pointID string, response interface{}, err error) bool {
if err != nil {
return true
}
// OPC UA特定的失败检查
// ...
return false
}
func (o *OPCUAPointAdapter) HandlePointSuccess(node *DeviceNodeTemplate, pointID string) {
resetPointFailure(node, pointID)
}
func (o *OPCUAPointAdapter) HandlePointFailure(node *DeviceNodeTemplate, pointID string) {
incrementPointFailure(node, pointID)
}
func (o *OPCUAPointAdapter) ShouldDegradePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldDegradePoint(pointID, node)
}
func (o *OPCUAPointAdapter) ShouldProbePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldProbePoint(pointID, node)
}
func (o *OPCUAPointAdapter) UpdateProbeInterval(node *DeviceNodeTemplate, pointID string) {
updateProbeInterval(pointID, node)
}
func (o *OPCUAPointAdapter) BuildBatches(points []model.Point) [][]model.Point {
// OPC UA特定的批次构建逻辑
var batches [][]model.Point
var currentBatch []model.Point
for _, point := range points {
if len(currentBatch) >= o.batchSize {
batches = append(batches, currentBatch)
currentBatch = []model.Point{point}
} else {
currentBatch = append(currentBatch, point)
}
}
if len(currentBatch) > 0 {
batches = append(batches, currentBatch)
}
return batches
}
// S7点位适配器
type S7PointAdapter struct {
batchSize int
}
func NewS7PointAdapter() *S7PointAdapter {
return &S7PointAdapter{batchSize: 80}
}
func (s *S7PointAdapter) ProtocolType() string {
return "s7"
}
func (s *S7PointAdapter) IsPointFailure(pointID string, response interface{}, err error) bool {
if err != nil {
return true
}
// S7特定的失败检查
// ...
return false
}
func (s *S7PointAdapter) HandlePointSuccess(node *DeviceNodeTemplate, pointID string) {
resetPointFailure(node, pointID)
}
func (s *S7PointAdapter) HandlePointFailure(node *DeviceNodeTemplate, pointID string) {
incrementPointFailure(node, pointID)
}
func (s *S7PointAdapter) ShouldDegradePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldDegradePoint(pointID, node)
}
func (s *S7PointAdapter) ShouldProbePoint(node *DeviceNodeTemplate, pointID string) bool {
return shouldProbePoint(pointID, node)
}
func (s *S7PointAdapter) UpdateProbeInterval(node *DeviceNodeTemplate, pointID string) {
updateProbeInterval(pointID, node)
}
func (s *S7PointAdapter) BuildBatches(points []model.Point) [][]model.Point {
// S7特定的批次构建逻辑
var batches [][]model.Point
var currentBatch []model.Point
for _, point := range points {
if len(currentBatch) >= s.batchSize {
batches = append(batches, currentBatch)
currentBatch = []model.Point{point}
} else {
currentBatch = append(currentBatch, point)
}
}
if len(currentBatch) > 0 {
batches = append(batches, currentBatch)
}
return batches
}
2.13.5.2 与设备画像集成
基于Modbus点位降级策略，我们优化了与设备画像的集成，直接从DeviceNodeTemplate中获取点位降级状态，这样可以更高效地更新设备画像。
// 在设备画像更新中集成点位降级状态
func (a *DeviceAdapterManager) UpdateProfile(deviceID string, responseTime time.Duration, err error, successCount, failCount int) {
// 现有逻辑...
// 集成点位降级状态
if stateManager != nil {
// 从设备节点获取点位状态
node := stateManager.GetNode(deviceID)
if node != nil {
degradedPointCount := 0
totalPointCount := 0
// 统计降级点位数量
if node.PointDegradedMap != nil {
for _, degraded := range node.PointDegradedMap {
totalPointCount++
if degraded {
degradedPointCount++
}
}
}
// 统计失败点位数量
failurePointCount := 0
if node.PointFailureMap != nil {
for _, failCount := range node.PointFailureMap {
if failCount > 0 {
failurePointCount++
}
}
}
// 更新设备画像中的点位状态
if totalPointCount > 0 {
profile.异常点位数量 = degradedPointCount
profile.故障点位数量 = failurePointCount
profile.点位降级率 = float64(degradedPointCount) / float64(totalPointCount)
}
// 更新设备稳定性评分
if totalPointCount > 0 {
stablePointCount := totalPointCount - degradedPointCount - failurePointCount
profile.稳定性评分 = float64(stablePointCount) / float64(totalPointCount)
}
}
}
// 现有逻辑...
}
// 在设备画像中添加点位降级相关字段
type DeviceProfile struct {
// 现有字段...
异常点位数量 int `json:"abnormal_point_count"`
故障点位数量 int `json:"failure_point_count"`
点位降级率 float64 `json:"point_degradation_rate"`
稳定性评分 float64 `json:"stability_score"`
// 其他字段...
}
2.13.6 监控与告警
基于Modbus点位降级策略，我们优化了监控与告警接口，直接从DeviceNodeTemplate中获取点位状态，提供更实时、准确的监控信息。
// 点位状态监控接口
type PointDegradationMonitor interface {
// 获取点位状态统计
GetPointStats() map[string]PointStats
// 获取降级点位列表
GetDegradedPoints() []PointStats
// 获取点位状态历史
GetPointStatusHistory(pointID, deviceID string, duration time.Duration) []PointStats
// 导出点位状态数据
ExportPointStatuses() ([]byte, error)
// 获取点位降级建议
GetDegradationSuggestions() []DegradationSuggestion
// 获取设备点位状态概览
GetDevicePointStatusOverview(deviceID string) DevicePointStatusOverview
}
// 点位统计信息
type PointStats struct {
PointID string `json:"point_id"`
DeviceID string `json:"device_id"`
Protocol string `json:"protocol"`
Status string `json:"status"` // normal, warning, degraded
FailureCount int `json:"failure_count"`
LastSuccessTime time.Time `json:"last_success_time"`
LastFailureTime time.Time `json:"last_failure_time"`
ProbeInterval int `json:"probe_interval"`
LastProbeTime time.Time `json:"last_probe_time"`
Degraded bool `json:"degraded"`
}
// 设备点位状态概览
type DevicePointStatusOverview struct {
DeviceID string `json:"device_id"`
TotalPoints int `json:"total_points"`
NormalPoints int `json:"normal_points"`
WarningPoints int `json:"warning_points"`
DegradedPoints int `json:"degraded_points"`
DegradationRate float64 `json:"degradation_rate"`
LastUpdated time.Time `json:"last_updated"`
}
// 降级建议
type DegradationSuggestion struct {
PointID string `json:"point_id"`
DeviceID string `json:"device_id"`
Protocol string `json:"protocol"`
FailureCount int `json:"failure_count"`
Status string `json:"status"`
Suggestion string `json:"suggestion"`
Priority string `json:"priority"` // high, medium, low
EstimatedTime string `json:"estimated_time"` // 预计恢复时间
}
// 监控实现
func (m *DeviceStateManager) GetPointStats() map[string]PointStats {
stats := make(map[string]PointStats)
for deviceID, node := range m.nodes {
if node.PointFailureMap != nil {
for pointID, failureCount := range node.PointFailureMap {
status := "normal"
degraded := false
if node.PointDegradedMap != nil && node.PointDegradedMap[pointID] {
status = "degraded"
degraded = true
} else if failureCount > 0 {
status = "warning"
}
lastSuccessTime := time.Time{}
if node.PointLastSuccessTime != nil {
lastSuccessTime = node.PointLastSuccessTime[pointID]
}
probeInterval := 8
lastProbeTime := time.Time{}
if node.PointProbeInterval != nil {
probeInterval = node.PointProbeInterval[pointID]
}
if node.PointLastProbeTime != nil {
lastProbeTime = node.PointLastProbeTime[pointID]
}
stats[pointID+"_"+deviceID] = PointStats{
PointID: pointID,
DeviceID: deviceID,
Protocol: node.ProtocolType, // 假设node中有协议类型字段
Status: status,
FailureCount: failureCount,
LastSuccessTime: lastSuccessTime,
LastFailureTime: time.Now(), // 可以从其他地方获取
ProbeInterval: probeInterval,
LastProbeTime: lastProbeTime,
Degraded: degraded,
}
}
}
}
return stats
}
// 获取降级点位列表
func (m *DeviceStateManager) GetDegradedPoints() []PointStats {
var degradedPoints []PointStats
stats := m.GetPointStats()
for _, stat := range stats {
if stat.Degraded {
degradedPoints = append(degradedPoints, stat)
}
}
return degradedPoints
}
// 获取设备点位状态概览
func (m *DeviceStateManager) GetDevicePointStatusOverview(deviceID string) DevicePointStatusOverview {
node := m.GetNode(deviceID)
if node == nil {
return DevicePointStatusOverview{
DeviceID: deviceID,
TotalPoints: 0,
NormalPoints: 0,
WarningPoints: 0,
DegradedPoints: 0,
DegradationRate: 0,
LastUpdated: time.Now(),
}
}
totalPoints := 0
normalPoints := 0
warningPoints := 0
degradedPoints := 0
if node.PointFailureMap != nil {
totalPoints = len(node.PointFailureMap)
for pointID, failureCount := range node.PointFailureMap {
if node.PointDegradedMap != nil && node.PointDegradedMap[pointID] {
degradedPoints++
} else if failureCount > 0 {
warningPoints++
} else {
normalPoints++
}
}
}
degradationRate := 0.0
if totalPoints > 0 {
degradationRate = float64(degradedPoints) / float64(totalPoints)
}
return DevicePointStatusOverview{
DeviceID: deviceID,
TotalPoints: totalPoints,
NormalPoints: normalPoints,
WarningPoints: warningPoints,
DegradedPoints: degradedPoints,
DegradationRate: degradationRate,
LastUpdated: time.Now(),
}
}
2.13.7 配置参数
基于Modbus点位降级策略，我们优化了配置参数，使其更符合实际的点位状态管理需求，支持从DeviceNodeTemplate中直接管理点位状态。
// 点位降级配置
type PointDegradationConfig struct {
EnablePointDegradation bool // 启用电位降级
FailureThreshold int // 失败次数阈值（默认3）
MaxProbeInterval int // 最大探测间隔（秒，默认1024）
InitialProbeInterval int // 初始探测间隔（秒，默认8）
ProbeMultiplier int // 探测间隔乘数（默认2）
StatusExpiry time.Duration // 状态过期时间（默认24小时）
// 协议特定配置
ProtocolConfigs map[string]ProtocolDegradationConfig
// 监控配置
EnableMonitoring bool // 启用监控
MonitoringInterval int // 监控间隔（秒，默认60）
EnableAlerting bool // 启用告警
AlertThreshold float64 // 告警阈值（默认0.3，即30%点位降级）
}
// 协议特定降级配置
type ProtocolDegradationConfig struct {
FailureThreshold int // 失败次数阈值
MaxProbeInterval int // 最大探测间隔（秒）
InitialProbeInterval int // 初始探测间隔（秒）
CustomParameters map[string]interface{} // 协议特定参数
// 协议特定的批量处理配置
BatchSize int // 批量处理大小
BatchTimeout int // 批量处理超时（毫秒）
}
// 设备节点配置扩展
type DeviceNodeConfig struct {
// 现有配置...
PointDegradation PointDegradationConfig // 点位降级配置
// 其他配置...
}
// 配置加载示例
func LoadPointDegradationConfig() PointDegradationConfig {
return PointDegradationConfig{
EnablePointDegradation: true,
FailureThreshold: 3,
MaxProbeInterval: 1024,
InitialProbeInterval: 8,
ProbeMultiplier: 2,
StatusExpiry: 24 * time.Hour,
EnableMonitoring: true,
MonitoringInterval: 60,
EnableAlerting: true,
AlertThreshold: 0.3,
ProtocolConfigs: map[string]ProtocolDegradationConfig{
"modbus": {
FailureThreshold: 3,
MaxProbeInterval: 1024,
InitialProbeInterval: 8,
BatchSize: 100,
BatchTimeout: 5000,
},
"bacnet": {
FailureThreshold: 3,
MaxProbeInterval: 1024,
InitialProbeInterval: 8,
BatchSize: 50,
BatchTimeout: 3000,
},
"opcua": {
FailureThreshold: 3,
MaxProbeInterval: 1024,
InitialProbeInterval: 8,
BatchSize: 200,
BatchTimeout: 10000,
},
"s7": {
FailureThreshold: 3,
MaxProbeInterval: 1024,
InitialProbeInterval: 8,
BatchSize: 80,
BatchTimeout: 4000,
},
},
}
}
// 设备适配器配置
type DeviceAdapterConfig struct {
EnableAdaptiveCollection bool // 启用自适应采集
MinResponseTime time.Duration // 最小响应时间
MaxResponseTime time.Duration // 最大响应时间
StabilityThreshold float64 // 稳定性阈值
BatchOptimization bool // 启用批量优化
ConcurrencyLimit int // 并发限制
ProfileExpiry time.Duration // 画像过期时间
// RTT相关配置
RTTSampleWindow int // RTT采样窗口大小
RTTAlphaGain float64 // EWMA增益系数
RTTThresholdHigh int64 // 高延迟阈值
RTTThresholdLow int64 // 低延迟阈值
// MTU相关配置
MTUMax int // 最大MTU
MTUMin int // 最小MTU
MTUStep int // MTU探测步长
MTUHysteresis float64 // MTU抖动滞后因子
// Gap相关配置
GapMaxHole int // 最大允许地址空洞
GapFillStrategy int // 空洞填充策略
GapDynamicEnable bool // 启用基于RTT的动态Gap调整
// 心跳相关配置
HeartbeatInterval int // 心跳间隔
HeartbeatSuppressEnable bool // 启用心跳抑制
}
2.12.2 监控接口
// 监控接口
type DeviceAdapterMonitor interface {
// 获取设备状态统计
GetDeviceStats() map[string]DeviceStats
// 获取协议性能统计
GetProtocolStats() map[string]ProtocolStats
// 获取异常设备列表
GetUnstableDevices() []string
// 导出设备画像数据
ExportProfiles() ([]byte, error)
// 获取采集优化建议
GetOptimizationSuggestions() []OptimizationSuggestion
// 获取RTT统计
GetRTTStats() map[string]RTTStats
// 获取MTU统计
GetMTUStats() map[string]MTUStats
// 获取Gap统计
GetGapStats() map[string]GapStats
}
// 设备统计信息
type DeviceStats struct {
DeviceID string `json:"device_id"`
Name string `json:"name"`
Protocol string `json:"protocol"`
Status string `json:"status"`
StabilityScore float64 `json:"stability_score"`
ResponseTime float64 `json:"response_time"` // 毫秒
ErrorRate float64 `json:"error_rate"`
采集成功率 float64 `json:"success_rate"`
连续失败次数 int `json:"consecutive_failures"`
LastUpdated time.Time `json:"last_updated"`
// RTT相关
EWMARTT int64 `json:"ewma_rtt"` // 微秒
// MTU相关
CurrentMTU int `json:"current_mtu"`
// Gap相关
CurrentGap int `json:"current_gap"`
}
// 协议统计信息
type ProtocolStats struct {
Protocol string `json:"protocol"`
DeviceCount int `json:"device_count"`
OnlineCount int `json:"online_count"`
AvgResponseTime float64 `json:"avg_response_time"`
AvgErrorRate float64 `json:"avg_error_rate"`
LastUpdated time.Time `json:"last_updated"`
// RTT相关
AvgEWMARTT int64 `json:"avg_ewma_rtt"`
// MTU相关
AvgMTU int `json:"avg_mtu"`
// Gap相关
AvgGap int `json:"avg_gap"`
}
// RTT统计信息
type RTTStats struct {
DeviceID string `json:"device_id"`
EWMARTT int64 `json:"ewma_rtt"`
MinRTT int64 `json:"min_rtt"`
MaxRTT int64 `json:"max_rtt"`
AvgRTT int64 `json:"avg_rtt"`
LastUpdated time.Time `json:"last_updated"`
}
// MTU统计信息
type MTUStats struct {
DeviceID string `json:"device_id"`
CurrentMTU int `json:"current_mtu"`
MaxMTU int `json:"max_mtu"`
MinMTU int `json:"min_mtu"`
LastNegotiation time.Time `json:"last_negotiation"`
}
// Gap统计信息
type GapStats struct {
DeviceID string `json:"device_id"`
CurrentGap int `json:"current_gap"`
MaxGap int `json:"max_gap"`
MergedRequests uint64 `json:"merged_requests"`
SavedRequests uint64 `json:"saved_requests"`
FillEfficiency float64 `json:"fill_efficiency"`
LastUpdated time.Time `json:"last_updated"`
}
// 优化建议
type OptimizationSuggestion struct {
DeviceID string `json:"device_id"`
Suggestion string `json:"suggestion"`
Priority string `json:"priority"` // high, medium, low
ExpectedImprovement float64 `json:"expected_improvement"`
}
3. 前端（嵌入式/设备侧）修改要点
为了配合网关的高效采集，下位机/传感器端需配合优化：
3.1 精准时序控制
- 中断优先级: 将通信接口 (UART/ETH/CAN) 中断优先级提至最高，确保 RTT 时间戳精度 < 1 ms。
- 硬件捕获: 使用 MCU 硬件定时器 Capture 通道捕获帧首/帧尾时间，消除软件抖动。
3.2 协议栈增强
- 心跳响应: 实现对网关
HeartbeatInterval探测包的快速响应 (Echo)。 - 大包支持: 确保存储区支持连续读写，配合网关的 Gap 合并策略。
3.3 内存管理
- 零拷贝: 采集数据队列使用 DMA + 环形缓冲，避免数据搬运。
- 动态 Buffer: 根据协商后的 MTU 动态分配/复用接收缓冲区，避免内存浪费。
4. 后端（Go 网关侧）修改要点
4.1 协议栈扩展
- 控制特征值: 增加
COLLECT_CTRL指令处理，支持通过 API 下发 RTT 采样周期与 MTU 策略。 - 参数持久化: 确保
HeartbeatInterval,BufferSize,Qos等参数正确保存至selfpara/*.json。
4.2 流量优化
- 心跳抑制: 在调度器中实现
ShouldSendHeartbeat(interval)逻辑，若time.Since(LastActivity) < interval则跳过心跳。 - ACK 压缩: 对高频 Qos=1 的采集包，实现 Bitmap ACK 或累积 ACK，减少反向确认包数量。
- 动态 Gap: 在协议驱动中，根据当前 MTU 与 RTT 动态计算最佳 Gap 值，平衡总线占用与响应速度。
4.3 运维支持
- 热更新: 提供 HTTP 接口
POST /api/v2/interface/communication/update_params支持参数 A/B 灰度测试。 - 增强日志: 日志中增加
seq_no,rtt_us,mtu_size,gap_val字段，便于大数据分析。
5. 关键步骤与里程碑
| 阶段 | 任务 | 验收标准 |
|---|---|---|
| Step-0 | 基准测试 | 输出当前延迟、丢包、功耗基线报告；确认现有协议配置加载正常。 |
| Step-1 | 高精打戳 | 驱动层实现纳秒/微秒级打戳；验证精度 < 1 ms。 |
| Step-2 | RTT 算法 | 实现 RTT 采样窗口与 EWMA 算法；验证在丢包场景下的曲线平滑度。 |
| Step-3 | MTU 状态机 | 实现二分协商状态机；单元测试覆盖率 100%。 |
| Step-4 | Gap 联动 | 实现 MTU -> Gap Max 的联动，以及 RTT -> Gap Current 的动态调整。 |
| Step-5 | 前后端联调 | 对接 API，前端 UI 实时展示 RTT 曲线、当前 MTU 及 Gap 合并效率。 |
| Step-6 | 压力测试 | 100k 采样点/秒持续 2h；延迟 99th < 10ms；内存无泄漏。 |
6. 实施计划
6.1 阶段一：核心模块开发
- 实现设备适配器核心接口
- 定义DeviceCommunicationProfile结构，包含RTT、MTU、Gap等字段
- 实现DeviceAdapter接口，添加RTT、MTU、Gap相关方法
- 实现ProtocolAdapter接口，添加协议特定的MTU和Gap计算
- 开发基础协议适配器
- Modbus协议适配器，实现RTT自适应和MTU协商
- BACnet协议适配器，实现COV订阅优化
- OPC UA协议适配器，实现订阅优化和批处理
- S7协议适配器，实现数据块优化和MTU协商
- 实现设备画像管理
- DeviceAdapterManager实现，集成RTT、MTU、Gap管理器
- 画像持久化存储
- 画像过期机制
- 实现RTT、MTU、Gap管理器
- RTTManager：实现RTT统计和EWMA计算
- MTUManager：实现MTU动态协商
- GapOptimizer：实现基于RTT和MTU的Gap优化
- 实现影子设备核心模块
- Real Shadow Store：纯内存运行时快照
- Virtual Shadow Engine：跨设备公式依赖计算
- Edge Compute Engine：差值/步长/滑动平均/阈值计算
- Atomic Dispatcher：版本推进 + QoS ACK 机制
- Shadow Ingress：批量写入 + QoS 0/1/2 + 去重
- 实现通用点位降级框架
- 扩展DeviceNodeTemplate结构，添加点位状态管理字段
- 实现批次构建函数，集成点位降级逻辑
- 开发指数级退避探测机制
- 与ChannelManager的collectDevice方法集成
- 与设备画像系统集成，同步点位降级状态
- 实现监控与告警接口
- 实现边缘运维与Gossip同步
- Gossip通讯模块：节点发现、成员心跳、增量反熵同步
- bbolt持久化层：配置存储、离线队列、冲突日志
- 冲突解决引擎：时间戳最新优先策略
- UUID管理工具：生成、校验、手工绑定、替换演练
6.2 阶段二：集成与测试
- 与现有采集模块集成
- 修改ChannelManager集成智能适配逻辑
- 驱动层参数优化，支持RTT、MTU、Gap参数
- 状态管理集成
- 心跳机制集成，实现智能业务心跳
- 集成点位降级框架，实现点位过滤和状态管理
- 进行多协议测试
- Modbus设备测试，验证RTT自适应和MTU协商
- BACnet设备测试，验证COV订阅优化
- OPC UA设备测试，验证订阅优化和批处理
- 混合协议测试，验证并发采集优化
- 多协议点位降级测试，验证不同协议的点位降级效果
- 优化并发采集性能
- 并发采集测试，验证采集顺序优化
- 批量采集测试，验证Gap合并策略
- 点位降级测试，验证降级和恢复机制
- 性能瓶颈分析，优化系统资源利用
6.3 阶段三：功能扩展
- 添加批量采集优化
- 同一总线设备批量采集
- 协议特定批量优化
- 批量采集效果评估
- 实现监控和统计功能
- 设备画像监控面板，展示RTT、MTU、Gap等指标
- 点位状态监控面板，展示点位降级状态和历史
- 协议性能统计
- 优化建议生成
- 实时监控API
- 支持更多协议类型
- 串口协议适配器
- 以太网IP协议适配器
- 三菱PLC协议适配器
- 欧姆龙PLC协议适配器
- 为新协议添加点位降级适配器
- 运维支持
- 热更新：提供HTTP接口支持参数A/B灰度测试
- 增强日志：日志中增加RTT、MTU、Gap、点位降级状态等字段
- 故障诊断：基于设备画像和点位降级状态的故障分析
- 点位降级配置管理：通过API调整点位降级参数
7. 预期效果
- 采集效率提升：通过智能适配、RTT优化、批量采集和点位降级，采集效率提升40%以上
- 系统稳定性提高：通过设备画像、状态预测和点位降级，系统稳定性提高30%
- 延迟优化：99th百分位延迟控制在10ms以内
- 吞吐量提升：通过MTU动态协商和Gap合并，吞吐量提升50%以上
- 资源利用率优化：合理的并发策略、批量采集和点位降级，减少系统资源占用
- 运维成本降低：自动化的设备管理、优化和点位降级，减少人工干预
- 可扩展性增强：模块化设计，易于添加新协议支持
- 点位故障隔离：通过点位降级，避免故障点位影响整体采集性能
- 故障恢复自动化：通过指数级退避探测，实现故障点位的自动恢复
8. 技术创新点
- 统一设备画像：为不同协议设备建立统一的画像结构，实现跨协议的设备管理
- 智能参数调整：基于设备历史数据自动调整采集参数，适应设备特性
- RTT自适应：基于往返时间动态调整采集参数，优化延迟和吞吐量
- MTU动态协商：智能调整最大传输单元，提高数据传输效率
- Gap合并优化：基于RTT和MTU动态调整Gap值，提高批量读取效率
- 并发优化策略：基于设备稳定性和响应时间的智能调度算法
- 智能业务心跳：根据业务通信自动调整心跳频率，减少网络流量
- 实时监控与分析：提供设备性能监控和优化建议
- 通用点位降级框架：为所有协议提供统一的点位降级机制，实现故障隔离
- 指数级退避探测：对降级点位采用智能探测机制，实现自动恢复
- 协议无关的点位管理：通过适配器模式，支持不同协议的点位状态管理
9. 总结
设备级别智能适配公共模块的设计，为多协议环境下的设备管理和采集优化提供了统一的解决方案。通过设备画像、智能调度、RTT自适应、MTU动态协商、批量优化和点位降级等技术，可以显著提高采集效率和系统稳定性，同时为未来的功能扩展和协议支持奠定基础。
该方案充分考虑了当前项目的架构特点，与现有系统无缝集成，同时引入了先进的智能适配机制，为边缘计算南向采集通道的性能优化提供了有力支持。
10. 测试验证方法与验收标准
10.1 测试验证方法
10.1.1 单元测试
- 点位状态管理测试
- 测试点位状态的初始化和更新
- 测试点位失败次数的统计
- 测试点位降级状态的切换
- 测试指数级退避探测机制
- 协议适配器测试
- 测试Modbus点位适配器的实现
- 测试BACnet点位适配器的实现
- 测试OPC UA点位适配器的实现
- 测试S7点位适配器的实现
- 点位降级管理器测试
- 测试点位过滤逻辑
- 测试批量点位处理
- 测试与设备画像的集成
- 测试并发访问安全性
10.1.2 集成测试
- 与ChannelManager集成测试
- 测试点位降级框架与ChannelManager的集成
- 测试采集流程中的点位过滤
- 测试点位状态更新的准确性
- 多协议集成测试
- 测试不同协议下的点位降级效果
- 测试混合协议场景下的点位管理
- 测试协议切换时的点位状态处理
- 性能测试
- 测试点位降级对采集性能的影响
- 测试大量点位场景下的性能表现
- 测试故障点位对系统的影响
- 故障注入测试
- 模拟点位故障场景
- 验证点位降级机制的有效性
- 验证故障恢复机制的可靠性
10.1.3 端到端测试
- 完整采集流程测试
- 测试从设备配置到数据采集的完整流程
- 验证点位降级在实际场景中的效果
- 测试系统稳定性和可靠性
- 故障恢复测试
- 模拟设备故障和恢复
- 验证点位自动恢复机制
- 测试系统的容错能力
- 边界场景测试
- 测试网络异常场景
- 测试设备离线场景
- 测试大量点位同时失败场景
10.2 验收标准
10.2.1 功能验收标准
- 点位降级功能
- 连续失败3次的点位应被降级
- 降级点位应采用指数级间隔探测
- 成功采集一次后应自动恢复
- 不同协议的点位应正确降级
- 性能验收标准
- 点位降级应减少30%以上的无效通信
- 系统整体采集效率应提升20%以上
- 99th百分位延迟应控制在10ms以内
- 支持至少1000个点位的同时管理
- 稳定性验收标准
- 系统应能稳定运行72小时以上
- 故障点位不应影响其他点位的采集
- 系统应能自动从故障中恢复
- 内存使用应保持稳定，无泄漏
- 可扩展性验收标准
- 应支持轻松添加新协议的点位适配器
- 配置参数应可通过API动态调整
- 应通过现有 diagnostics API 与结构化日志提供可观测性
- 应提供详细的日志和监控信息
10.2.2 测试通过标准
- 单元测试覆盖率
- 核心代码覆盖率应达到80%以上
- 关键路径覆盖率应达到100%
- 集成测试通过标准
- 所有集成测试用例应通过
- 无严重或中等级别缺陷
- 性能测试通过标准
- 性能指标应达到预期目标
- 系统应能处理预期的负载
- 端到端测试通过标准
- 完整流程测试应通过
- 故障恢复测试应通过
- 边界场景测试应通过
11. 交付物
- 设计文档:
docs/[TODO]边缘计算南向采集优化方案2026第二季度.md(本文档) - 架构总览:
docs/edge/边缘网关架构设计总览.md - ScanEngine 测试报告:
docs/TODO/ScanEngine重构测试报告.md - 接口代码（已实现）:
internal/core/device_adapter.gointernal/core/rtt_manager.go,mtu_manager.go,gap_optimizer.gointernal/core/shadow_core.go,shadow_ingress.go,shadow_device_optimizer.gointernal/core/scan_engine.go,scan_engine_compat.go,execution_layer.gointernal/core/virtual_shadow_engine.gointernal/driver/modbus/modbus_executor.go
- 接口代码（Q3 计划，原 §11 未交付）:
internal/core/point_degradation_manager.go❌internal/core/gossip_sync.go❌internal/core/protocol_point_adapters.go❌
- 单元测试（已实现）:
internal/core/device_adapter_test.gointernal/core/shadow_core_test.go,shadow_ingress_test.gointernal/core/rtt_manager_test.go,mtu_manager_test.go,gap_optimizer_test.gointernal/core/scan_engine_test.go,scan_engine_large_scale_test.go
-
配置示例（未单独拆分 JSON，运行时写入 config.db）
12. Q2 实现审查（截至 2026-06-25）
对照 §5 里程碑、§6 实施计划与代码库实际状态。状态：✅ 已完成 🟡 部分完成 ❌ 未实现 🗑️ 已替代
12.1 里程碑 Step-0 ~ Step-6 对照
| 阶段 | 计划内容 | 状态 | 实际交付 / 差距 |
|---|---|---|---|
| Step-0 | 基准测试、协议配置加载 | 🟡 | ScanEngine 压力测试报告（100 设备混合协议）；无正式延迟/功耗基线报告 |
| Step-1 | 驱动纳秒/微秒打戳 | 🟡 | OPC UA stampCollectionTime 使用本地采集时间；无硬件级 Capture |
| Step-2 | RTT 采样 + EWMA | ✅ | rtt_manager.go + 单元测试；ScanEngine 主路径未调用 UpdateRTT |
| Step-3 | MTU 二分协商状态机 | 🟡 | mtu_manager.go 有 NegotiateMTU；非完整 FSM，未接入驱动读路径 |
| Step-4 | MTU→Gap 联动 | 🟡 | gap_optimizer.go + ShadowDeviceOptimizer 联动；ExecutionLayer 读点前未消费 Gap |
| Step-5 | 前后端联调 RTT/MTU/Gap UI | ❌ | 画像字段写入 Shadow CommunicationProfile；UI 无 RTT 曲线/MTU/Gap 面板 |
| Step-6 | 100k 点/秒 · 2h · P99<10ms | ❌ | 大规模测试为 mock 驱动、短时；未达工业压测指标 |
12.2 §6.1 阶段一：核心模块 — 实现矩阵
| 模块 | 计划 | 状态 | 说明 |
|---|---|---|---|
| DeviceCommunicationProfile | 统一画像结构 | ✅ | model/types.go + Shadow 通信画像 |
| DeviceAdapter / ProtocolAdapter | 接口 + 实现 | 🟡 | device_adapter.go 存在；ProtocolAdapterRegistry 未全面接入驱动 |
| DeviceAdapterManager | 画像管理 | 🟡 | 已创建；画像未持久化到 DB，无过期机制 |
| RTTManager | EWMA | ✅ | 实现 + 测试 |
| MTUManager | 动态协商 | 🟡 | 简化算法；未驱动 Modbus PDU 分片 |
| GapOptimizer | 地址合并 | 🟡 | 算法存在；ModbusExecutor 未读 GapOptimizer 输出 |
| ShadowCore | 纯内存 | ✅ | shadow_core.go；Subscribe → WebSocket |
| ShadowIngress | QoS 批量写入 | 🟡 | 代码 + 测试；main.go 未挂载，ScanEngine 直写 Shadow |
| VirtualShadowEngine | 公式依赖 | 🟡 | virtual_shadow_engine.go；Subscribe 已接 Shadow |
| EdgeComputeManager | 阈值/窗口/表达式 | ✅ | 规则 CRUD + pipeline 触发 |
| Atomic Dispatcher / QoS ACK | CAS + ACK | 🟡 | CompareAndSwap 存在；QoS ACK 链未产品化 |
| 点位降级框架 | point_degradation_manager | ❌ | 文件未创建 |
| Gossip / bbolt 联机 | gossip_sync.go | ❌ | 未实现；SyncManager 在 main 中禁用 |
| CollectionScheduler | 画像驱动调度 | 🗑️ | 已由 ScanEngine 替代（collection_scheduler.go 已删） |
| CommDriver HAL | 统一物理层 | ❌ | 未实现 |
| 零拷贝 Ring Buffer | 缓存层 | ❌ | 未实现 |
| Traffic-Aware Heartbeat | 智能心跳 | ❌ | 未实现 |
12.3 §6.2 阶段二：集成 — 实现矩阵
| 集成项 | 状态 | 说明 |
|---|---|---|
| ChannelManager + 智能适配 | ✅ | ScanEngineAdapter 注册任务；deviceLoop / collectDevice 已移除 |
| 驱动层 RTT/MTU/Gap 参数 | 🟡 | ModbusExecutor + ConnectionController 新增；Gap 未闭环 |
| 状态管理 FinalizeCollect | ✅ | ScanEngine 经 finalizeScanCollect → FinalizeCollect 回写设备状态 |
| 心跳机制 | ❌ | 未实现 |
| 点位降级集成 | ❌ | 未实现 |
| 多协议实测报告 | 🟡 | OPC UA 点位时间戳问题已修；缺 Modbus/BACnet 实测文档 |
12.4 §6.3 阶段三：扩展 — 实现矩阵
| 扩展项 | 状态 | 说明 |
|---|---|---|
| 同总线批量采集 | 🟡 | ModbusExecutor PointScheduler；未与 GapOptimizer 联动 |
| 监控面板 RTT/MTU/Gap | ❌ | 南向指标卡片有部分 metrics；无画像专项 UI |
| DLT645 / Omron / Mitsubishi | ✅ | 驱动已注册 |
| EtherNet/IP | ✅ | 驱动 + Limited 执行模式 |
热更新 API /api/v2/... |
❌ | 配置热更新走标准 CRUD |
| 增强日志 seq/rtt/mtu/gap | ❌ | 结构化字段未统一 |
| 优化建议 API | ❌ | GetOptimizationSuggestions 未暴露 HTTP |
12.5 架构级变更（Q2 外溢，2026-06 已完成）
以下不在原 V2 正文但已落地，影响 Q3 规划基线：
| 变更 | 文件 | 意义 |
|---|---|---|
| ScanEngine 统一调度 | scan_engine.go + compat |
替代 per-device goroutine loop |
| ExecutionLayer 三模式 | serial / parallel / limited | 对标 Kepware 通道隔离 |
| 防饿死 + 背压 + 资源上限 | anti-starvation 300s, backpressure 512 | 工业调度骨架 |
| DB 单源配置 | config.db + runtime.db + ConfigStore |
安装向导 + 配置 DB 强一致写入 |
| ShadowCore 接入 ScanEngine | main.go SetShadowCore |
采集 → 影子 → WebSocket |
| UI 读影子 | WS Subscribe + GetDevicePoints 优先影子 | collected_at / updated_at 双时间戳 |
| 安装模式 | Install.vue + install_handler | 无 DB 首启 |
12.6 当前运行时数据流（实际）
ScanEngine → Driver.ReadPoints → ShadowCore.WriteShadowDevice → WebSocket
↓ (断层)
DataPipeline ✗
↓
EdgeCompute / Northbound / values 历史
核心断层：周期采集未进入 DataPipeline，边缘规则与北向遥测与 UI 影子 不同源。
12.7 Q2 目标达成度评估
| Q2 预期效果 | 达成度 | 说明 |
|---|---|---|
| 采集效率 +40% | 🟡 未量化 | 调度统一有收益；块读/Gap 未闭环 |
| 稳定性 +30% | 🟡 | ScanEngine 防饿死有效；点位降级缺失 |
| P99 延迟 <10ms | ❌ | 无生产压测证据 |
| 吞吐 +50% | ❌ | MTU/Gap 未驱动 I/O |
| 点位故障隔离 | ❌ | 降级框架未做 |
| 72h 稳定运行 | ❌ | 未执行 |
Q2 结论：调度层（ScanEngine）+ 影子层（ShadowCore）+ 画像算法模块（RTT/MTU/Gap） 主体代码已交付，但 集成闭环不足，距离方案 V2 描述的端到端智能采集仍有明显差距。Q3 重点应从「模块存在」转向「Kepware 式产品能力闭环」。
13. 2026 Q3 规划 — 对标 Kepware 路线
完整 Q3 方案已独立成文 → 边缘计算南向采集优化方案2026第三季度.md
以下 §13–§14 为 Q2 文档内保留摘要；实施、验收、进度跟踪以 Q3 文档为准。
参考 KEPServerEX 核心概念：Tag Database · Scan Mode/Class · Block I/O · Publish Rate · Store & Forward · Diagnostics。
详细生命周期见：docs/edge/边缘网关架构设计总览.md
13.1 Kepware 概念 → EdgeX 映射
| Kepware 概念 | 含义 | EdgeX Q3 目标模块 |
|---|---|---|
| Tag Database | 统一 Tag 命名、属性、缩放、EU | TagRegistry（DB 扩展或 Devices 之上抽象层） |
| Scan Mode | 同步读 / 订阅 / 只读一次 | 驱动 Capability + ScanEngine 任务类型 |
| Scan Class | 100ms / 1s / 10s 分组调度 | ScanEngine 多 Interval 或点位分组任务 |
| Block I/O | 寄存器块读、减少报文 | Modbus GapOptimizer → ExecutionLayer 分片 |
| Publish Rate | 采集速率 ≠ 上报速率 | Shadow 全采 + Pipeline 限频推送北向 |
| Store & Forward | 断网缓存补发 | 统一 OfflineQueue（南向 values + 北向 cache） |
| Diagnostics | 通道/设备/Tag 级延迟与错误 | GET /diagnostics/scan-engine 与 /api/diagnostics/* |
| Media Level Redundancy | 主备通道 | Phase Q3-B（设计预留，不强制实现） |
13.2 Q3 总目标
- 统一数据面：Shadow 为 SoT，Pipeline 为扇出，北向/边缘/历史与 UI 一致。
- 可预测调度：Scan Class + 调度 SLA 可观测，工业模式可关闭失败退避。
- 高吞吐 I/O：Modbus/OPC UA 块读闭环，万 Tag 压测出基线报告。
- 点位级韧性：Tag 降级与探测恢复（补 Q2 缺失）。
- 可运维：Diagnostics 对标 Kepware Event Log / Statistics。
13.3 Q3 分期计划
Q3-A（7 月）：统一数据面 + 调度可观测 — P0
| 编号 | 任务 | 交付 | 验收 |
|---|---|---|---|
| A1 | Shadow → Pipeline 桥接 | ShadowCore.Subscribe 内 Push model.Value |
周期采集触发 ECM + NBM + values |
| A2 | ScanEngine 状态回写 | executeTaskAsync → FinalizeCollect |
设备在线率与采集成功率一致 |
| A3 | 删除/归档 collectDevice |
清理死代码 | 无重复采集路径 |
| A4 | ShadowIngress 评估 | 决定保留直连或统一 Ingress | 文档化单一写入规范 |
| A5 | 调度指标 | scan_lag_ms / starvation_rescue_total / task_overdue |
GET /diagnostics/scan-engine 与 sla_warnings[] |
| A6 | 集成测试 | shadow_pipeline_integration_test.go |
四路（WS/规则/北向/DB）时间戳一致 |
Kepware 对齐：Publish Rate 分离（A1 北向可独立限频）；Diagnostics 基础（A5）。
Q3-B（8 月）：Scan Class + 块读闭环 — P0
| 编号 | 任务 | 交付 | 验收 |
|---|---|---|---|
| B1 | Scan Class 模型 | 设备/点位 scan_class: fast\|normal\|slow |
同设备多周期并存 |
| B2 | 工业调度模式 | degrade_on_failure: false |
失败仅 Bad，不拉长 Interval |
| B3 | Event-driven 调度 | 堆顶 timer + 10ms 兜底 | CPU 占用下降（对比报告） |
| B4 | Gap → Modbus 读路径 | ExecutionLayer 读前分片 | 连续寄存器合并读，请求数下降 ≥30% |
| B5 | MTU → 分片上限 | ModbusExecutor 读 PDU 用 CurrentMTU | 单帧不超协商值 |
| B6 | ScanEngine → RTT | 每次 Execute 后 UpdateDeviceRTT |
画像 EWMARTT 随采集更新 |
| B7 | OPC UA 分批 Read | 超 N 点拆分 + MaxAge | 500+ 点设备稳定 |
Kepware 对齐：Scan Class + Block I/O。
Q3-C（9 月）：Tag 韧性 + 压测 + Diagnostics — P1
| 编号 | 任务 | 交付 | 验收 |
|---|---|---|---|
| C1 | 点位降级框架 | point_degradation_manager.go |
连续 3 次失败降级，指数探测恢复 |
| C2 | Tag 级 Quality | ShadowPoint 保留 Bad + 降级标记 | UI 显示降级状态 |
| C3 | 万 Tag 压测 | 报告：吞吐、P50/P99 调度延迟、内存 | 1w Tag @1s，2h 无泄漏 |
| C4 | Diagnostics API | 通道/设备：RTT、MTU、Gap、成功率 | 对标 Kepware Statistics 页 |
| C5 | UI 画像面板 | DeviceMetricsCard 扩展 RTT/Gap | Step-5 补完 |
| C6 | Store & Forward 统一 | 北向断线 + 南向 history 策略文档化并实现 | 恢复后补发 |
Kepware 对齐：Tag 级 Diagnostics + Store & Forward。
13.4 Q3 明确不做 / 推迟（Q4+）
| 项 | 原因 |
|---|---|
| Gossip 多节点联机 | Sync 曾致路径污染，需独立 redesign |
| CommDriver HAL / Ring Buffer | 收益低于 ScanEngine + 驱动直连 |
| 通道级冗余 Failover | 依赖联机与驱动抽象成熟 |
| 完整 Atomic QoS2 | 先保证 QoS0 扇出一致 |
| 100k 点/秒 | Q3 目标 1w Tag 基线，10w 为 Q4 |
13.5 Q3 架构目标态
┌──────────────── Tag Database (scan_class, EU, alias) ────────────────┐
└───────────────────────────────┬────────────────────────────────────┘
│
┌───────────▼───────────┐
│ ScanEngine │
│ ScanClass · Priority │
│ Anti-starvation · Metrics│
└───────────┬───────────┘
│ Block I/O (Gap/MTU)
┌───────────▼───────────┐
│ ShadowCore (SoT) │
│ collected_at/updated_at│
└───────────┬───────────┘
│ Subscribe 扇出
┌──────────────────────┼──────────────────────┐
│ │ │
WebSocket DataPipeline VirtualShadow
│
┌─────────┴─────────┐
│ ECM · NBM · Hist │
│ (Publish Rate 限频) │
└───────────────────┘
13.6 Q3 验收标准（对标 Kepware 工业可用）
| 指标 | Q3 目标 |
|---|---|
| 数据一致性 | 同一 Tag 在 UI / MQTT / 规则 / values 四路值与 Quality 一致 |
| 调度 SLA | 1s Scan Class：95% 采集落在 [1s, 1s+100ms] |
| 防饿死 | 任意设备 5min 内至少被 rescue 或成功采集一次 |
| Modbus 块读 | 100 点离散地址 → 请求数减少 ≥30% |
| 点位降级 | 故障 Tag 不拖死同设备其他 Tag |
| 压测 | 1w Tag · 2h · 内存漂移 <5% · 无 panic |
| 可观测 | 每设备可查 EWMARTT、CurrentMTU、CurrentGap、成功率 |
13.7 Q3 交付物清单
| 交付物 | 路径 / 说明 |
|---|---|
| 统一数据面桥接 | shadow_core.go 或 shadow_pipeline_bridge.go |
| Scan Class 配置 | model.Device / model.Point 字段 + 迁移 |
| 点位降级 | internal/core/point_degradation_manager.go |
| 压测报告 | docs/testing/Q3_10k_tag_benchmark_2026Q3.md |
| Diagnostics API | internal/server/diagnostics_handler.go |
| 架构总览更新 | docs/edge/边缘网关架构设计总览.md §7 Roadmap |
| Q3 审查报告 | 本文 §14（Q3 末填写） |
13.8 风险与依赖
| 风险 | 缓解 |
|---|---|
| Shadow→Pipeline 双写性能 | 批量 Push；Pipeline 合并同设备消息 |
| Scan Class 任务数膨胀 | 按 Class 聚合为每设备每 Class 一任务，非每 Tag 一任务 |
| Modbus 双驱动注册 | 统一 ModbusExecutor，废弃 modbus.go 重复 init |
| 边缘规则延迟 | ECM 索引已 O(1)；扇通后需测规则触发延迟 <100ms |
| Q2 文档交付物路径过时 | §11 参考实现中 gossip/point_degradation 改为 Q3 计划项 |
14. Q3 末审查占位
计划填写时间：2026-09-30
填写内容：Q3-A/B/C 完成率、压测数据、与 Kepware 对标差距、Q4 规划入口。
| 项 | 计划 | 实际 | 备注 |
|---|---|---|---|
| A1 Shadow→Pipeline | |||
| B4 Gap 闭环 | |||
| C3 万 Tag 压测 | |||
| C1 点位降级 |
文档版本：V2.1 | 2026-06-25 增补 §12 Q2 审查 + §13 2026Q3 Kepware 路线规划
- edgex\config\gossip_sync_config.json
- edgex\config\point_degradation_config.json